Add missing hass type hint in component tests (m) (#124222)

This commit is contained in:
epenet 2024-08-19 15:36:37 +02:00 committed by GitHub
parent 02139fcca6
commit 5470d14a11
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 128 additions and 57 deletions

View File

@ -16,6 +16,7 @@ from homeassistant.components.media_player import (
SERVICE_CLEAR_PLAYLIST, SERVICE_CLEAR_PLAYLIST,
SERVICE_PLAY_MEDIA, SERVICE_PLAY_MEDIA,
SERVICE_SELECT_SOURCE, SERVICE_SELECT_SOURCE,
MediaPlayerEnqueue,
) )
from homeassistant.const import ( from homeassistant.const import (
ATTR_ENTITY_ID, ATTR_ENTITY_ID,
@ -35,70 +36,79 @@ from homeassistant.const import (
SERVICE_VOLUME_SET, SERVICE_VOLUME_SET,
SERVICE_VOLUME_UP, SERVICE_VOLUME_UP,
) )
from homeassistant.core import HomeAssistant
from homeassistant.loader import bind_hass from homeassistant.loader import bind_hass
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 on specified media player or all.""" """Turn on specified media player or all."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
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_on(hass, entity_id=ENTITY_MATCH_ALL): def turn_on(hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL) -> None:
"""Turn on specified media player or all.""" """Turn on specified media player or all."""
hass.add_job(async_turn_on, hass, entity_id) hass.add_job(async_turn_on, 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 off specified media player or all.""" """Turn off specified media player or all."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call(DOMAIN, SERVICE_TURN_OFF, data, blocking=True) await hass.services.async_call(DOMAIN, SERVICE_TURN_OFF, 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 off specified media player or all.""" """Turn off specified media player or all."""
hass.add_job(async_turn_off, hass, entity_id) hass.add_job(async_turn_off, hass, entity_id)
async def async_toggle(hass, entity_id=ENTITY_MATCH_ALL): async def async_toggle(hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL) -> None:
"""Toggle specified media player or all.""" """Toggle specified media player or all."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call(DOMAIN, SERVICE_TOGGLE, data, blocking=True) await hass.services.async_call(DOMAIN, SERVICE_TOGGLE, data, blocking=True)
@bind_hass @bind_hass
def toggle(hass, entity_id=ENTITY_MATCH_ALL): def toggle(hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL) -> None:
"""Toggle specified media player or all.""" """Toggle specified media player or all."""
hass.add_job(async_toggle, hass, entity_id) hass.add_job(async_toggle, hass, entity_id)
async def async_volume_up(hass, entity_id=ENTITY_MATCH_ALL): async def async_volume_up(
hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command for volume up.""" """Send the media player the command for volume up."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call(DOMAIN, SERVICE_VOLUME_UP, data, blocking=True) await hass.services.async_call(DOMAIN, SERVICE_VOLUME_UP, data, blocking=True)
@bind_hass @bind_hass
def volume_up(hass, entity_id=ENTITY_MATCH_ALL): def volume_up(hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL) -> None:
"""Send the media player the command for volume up.""" """Send the media player the command for volume up."""
hass.add_job(async_volume_up, hass, entity_id) hass.add_job(async_volume_up, hass, entity_id)
async def async_volume_down(hass, entity_id=ENTITY_MATCH_ALL): async def async_volume_down(
hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command for volume down.""" """Send the media player the command for volume down."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call(DOMAIN, SERVICE_VOLUME_DOWN, data, blocking=True) await hass.services.async_call(DOMAIN, SERVICE_VOLUME_DOWN, data, blocking=True)
@bind_hass @bind_hass
def volume_down(hass, entity_id=ENTITY_MATCH_ALL): def volume_down(hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL) -> None:
"""Send the media player the command for volume down.""" """Send the media player the command for volume down."""
hass.add_job(async_volume_down, hass, entity_id) hass.add_job(async_volume_down, hass, entity_id)
async def async_mute_volume(hass, mute, entity_id=ENTITY_MATCH_ALL): async def async_mute_volume(
hass: HomeAssistant, mute: bool, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command for muting the volume.""" """Send the media player the command for muting the volume."""
data = {ATTR_MEDIA_VOLUME_MUTED: mute} data = {ATTR_MEDIA_VOLUME_MUTED: mute}
@ -109,12 +119,16 @@ async def async_mute_volume(hass, mute, entity_id=ENTITY_MATCH_ALL):
@bind_hass @bind_hass
def mute_volume(hass, mute, entity_id=ENTITY_MATCH_ALL): def mute_volume(
hass: HomeAssistant, mute: bool, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command for muting the volume.""" """Send the media player the command for muting the volume."""
hass.add_job(async_mute_volume, hass, mute, entity_id) hass.add_job(async_mute_volume, hass, mute, entity_id)
async def async_set_volume_level(hass, volume, entity_id=ENTITY_MATCH_ALL): async def async_set_volume_level(
hass: HomeAssistant, volume: float, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command for setting the volume.""" """Send the media player the command for setting the volume."""
data = {ATTR_MEDIA_VOLUME_LEVEL: volume} data = {ATTR_MEDIA_VOLUME_LEVEL: volume}
@ -125,12 +139,16 @@ async def async_set_volume_level(hass, volume, entity_id=ENTITY_MATCH_ALL):
@bind_hass @bind_hass
def set_volume_level(hass, volume, entity_id=ENTITY_MATCH_ALL): def set_volume_level(
hass: HomeAssistant, volume: float, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command for setting the volume.""" """Send the media player the command for setting the volume."""
hass.add_job(async_set_volume_level, hass, volume, entity_id) hass.add_job(async_set_volume_level, hass, volume, entity_id)
async def async_media_play_pause(hass, entity_id=ENTITY_MATCH_ALL): async def async_media_play_pause(
hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command for play/pause.""" """Send the media player the command for play/pause."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call( await hass.services.async_call(
@ -139,48 +157,56 @@ async def async_media_play_pause(hass, entity_id=ENTITY_MATCH_ALL):
@bind_hass @bind_hass
def media_play_pause(hass, entity_id=ENTITY_MATCH_ALL): def media_play_pause(hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL) -> None:
"""Send the media player the command for play/pause.""" """Send the media player the command for play/pause."""
hass.add_job(async_media_play_pause, hass, entity_id) hass.add_job(async_media_play_pause, hass, entity_id)
async def async_media_play(hass, entity_id=ENTITY_MATCH_ALL): async def async_media_play(
hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command for play/pause.""" """Send the media player the command for play/pause."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call(DOMAIN, SERVICE_MEDIA_PLAY, data, blocking=True) await hass.services.async_call(DOMAIN, SERVICE_MEDIA_PLAY, data, blocking=True)
@bind_hass @bind_hass
def media_play(hass, entity_id=ENTITY_MATCH_ALL): def media_play(hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL) -> None:
"""Send the media player the command for play/pause.""" """Send the media player the command for play/pause."""
hass.add_job(async_media_play, hass, entity_id) hass.add_job(async_media_play, hass, entity_id)
async def async_media_pause(hass, entity_id=ENTITY_MATCH_ALL): async def async_media_pause(
hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command for pause.""" """Send the media player the command for pause."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call(DOMAIN, SERVICE_MEDIA_PAUSE, data, blocking=True) await hass.services.async_call(DOMAIN, SERVICE_MEDIA_PAUSE, data, blocking=True)
@bind_hass @bind_hass
def media_pause(hass, entity_id=ENTITY_MATCH_ALL): def media_pause(hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL) -> None:
"""Send the media player the command for pause.""" """Send the media player the command for pause."""
hass.add_job(async_media_pause, hass, entity_id) hass.add_job(async_media_pause, hass, entity_id)
async def async_media_stop(hass, entity_id=ENTITY_MATCH_ALL): async def async_media_stop(
hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command for stop.""" """Send the media player the command for stop."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call(DOMAIN, SERVICE_MEDIA_STOP, data, blocking=True) await hass.services.async_call(DOMAIN, SERVICE_MEDIA_STOP, data, blocking=True)
@bind_hass @bind_hass
def media_stop(hass, entity_id=ENTITY_MATCH_ALL): def media_stop(hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL) -> None:
"""Send the media player the command for stop.""" """Send the media player the command for stop."""
hass.add_job(async_media_stop, hass, entity_id) hass.add_job(async_media_stop, hass, entity_id)
async def async_media_next_track(hass, entity_id=ENTITY_MATCH_ALL): async def async_media_next_track(
hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command for next track.""" """Send the media player the command for next track."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call( await hass.services.async_call(
@ -189,12 +215,14 @@ async def async_media_next_track(hass, entity_id=ENTITY_MATCH_ALL):
@bind_hass @bind_hass
def media_next_track(hass, entity_id=ENTITY_MATCH_ALL): def media_next_track(hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL) -> None:
"""Send the media player the command for next track.""" """Send the media player the command for next track."""
hass.add_job(async_media_next_track, hass, entity_id) hass.add_job(async_media_next_track, hass, entity_id)
async def async_media_previous_track(hass, entity_id=ENTITY_MATCH_ALL): async def async_media_previous_track(
hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command for prev track.""" """Send the media player the command for prev track."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call( await hass.services.async_call(
@ -203,12 +231,16 @@ async def async_media_previous_track(hass, entity_id=ENTITY_MATCH_ALL):
@bind_hass @bind_hass
def media_previous_track(hass, entity_id=ENTITY_MATCH_ALL): def media_previous_track(
hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command for prev track.""" """Send the media player the command for prev track."""
hass.add_job(async_media_previous_track, hass, entity_id) hass.add_job(async_media_previous_track, hass, entity_id)
async def async_media_seek(hass, position, entity_id=ENTITY_MATCH_ALL): async def async_media_seek(
hass: HomeAssistant, position: float, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command to seek in current playing media.""" """Send the media player the command to seek in current playing media."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
data[ATTR_MEDIA_SEEK_POSITION] = position data[ATTR_MEDIA_SEEK_POSITION] = position
@ -216,14 +248,20 @@ async def async_media_seek(hass, position, entity_id=ENTITY_MATCH_ALL):
@bind_hass @bind_hass
def media_seek(hass, position, entity_id=ENTITY_MATCH_ALL): def media_seek(
hass: HomeAssistant, position: float, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command to seek in current playing media.""" """Send the media player the command to seek in current playing media."""
hass.add_job(async_media_seek, hass, position, entity_id) hass.add_job(async_media_seek, hass, position, entity_id)
async def async_play_media( async def async_play_media(
hass, media_type, media_id, entity_id=ENTITY_MATCH_ALL, enqueue=None hass: HomeAssistant,
): media_type: str,
media_id: str,
entity_id: str = ENTITY_MATCH_ALL,
enqueue: MediaPlayerEnqueue | bool | None = None,
) -> None:
"""Send the media player the command for playing media.""" """Send the media player the command for playing media."""
data = {ATTR_MEDIA_CONTENT_TYPE: media_type, ATTR_MEDIA_CONTENT_ID: media_id} data = {ATTR_MEDIA_CONTENT_TYPE: media_type, ATTR_MEDIA_CONTENT_ID: media_id}
@ -237,12 +275,20 @@ async def async_play_media(
@bind_hass @bind_hass
def play_media(hass, media_type, media_id, entity_id=ENTITY_MATCH_ALL, enqueue=None): def play_media(
hass: HomeAssistant,
media_type: str,
media_id: str,
entity_id: str = ENTITY_MATCH_ALL,
enqueue: MediaPlayerEnqueue | bool | None = None,
) -> None:
"""Send the media player the command for playing media.""" """Send the media player the command for playing media."""
hass.add_job(async_play_media, hass, media_type, media_id, entity_id, enqueue) hass.add_job(async_play_media, hass, media_type, media_id, entity_id, enqueue)
async def async_select_source(hass, source, entity_id=ENTITY_MATCH_ALL): async def async_select_source(
hass: HomeAssistant, source: str, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command to select input source.""" """Send the media player the command to select input source."""
data = {ATTR_INPUT_SOURCE: source} data = {ATTR_INPUT_SOURCE: source}
@ -253,18 +299,22 @@ async def async_select_source(hass, source, entity_id=ENTITY_MATCH_ALL):
@bind_hass @bind_hass
def select_source(hass, source, entity_id=ENTITY_MATCH_ALL): def select_source(
hass: HomeAssistant, source: str, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command to select input source.""" """Send the media player the command to select input source."""
hass.add_job(async_select_source, hass, source, entity_id) hass.add_job(async_select_source, hass, source, entity_id)
async def async_clear_playlist(hass, entity_id=ENTITY_MATCH_ALL): async def async_clear_playlist(
hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL
) -> None:
"""Send the media player the command for clear playlist.""" """Send the media player the command for clear playlist."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call(DOMAIN, SERVICE_CLEAR_PLAYLIST, data, blocking=True) await hass.services.async_call(DOMAIN, SERVICE_CLEAR_PLAYLIST, data, blocking=True)
@bind_hass @bind_hass
def clear_playlist(hass, entity_id=ENTITY_MATCH_ALL): def clear_playlist(hass: HomeAssistant, entity_id: str = ENTITY_MATCH_ALL) -> None:
"""Send the media player the command for clear playlist.""" """Send the media player the command for clear playlist."""
hass.add_job(async_clear_playlist, hass, entity_id) hass.add_job(async_clear_playlist, hass, entity_id)

View File

@ -31,7 +31,7 @@ async def setup_homeassistant(hass: HomeAssistant):
await async_setup_component(hass, "homeassistant", {}) await async_setup_component(hass, "homeassistant", {})
def create_group(hass, name): def create_group(hass: HomeAssistant, name: str) -> None:
"""Create a new person group. """Create a new person group.
This is a legacy helper method. Do not use it for new tests. This is a legacy helper method. Do not use it for new tests.
@ -40,7 +40,7 @@ def create_group(hass, name):
hass.async_create_task(hass.services.async_call(DOMAIN, SERVICE_CREATE_GROUP, data)) hass.async_create_task(hass.services.async_call(DOMAIN, SERVICE_CREATE_GROUP, data))
def delete_group(hass, name): def delete_group(hass: HomeAssistant, name: str) -> None:
"""Delete a person group. """Delete a person group.
This is a legacy helper method. Do not use it for new tests. This is a legacy helper method. Do not use it for new tests.
@ -49,7 +49,7 @@ def delete_group(hass, name):
hass.async_create_task(hass.services.async_call(DOMAIN, SERVICE_DELETE_GROUP, data)) hass.async_create_task(hass.services.async_call(DOMAIN, SERVICE_DELETE_GROUP, data))
def train_group(hass, group): def train_group(hass: HomeAssistant, group: str) -> None:
"""Train a person group. """Train a person group.
This is a legacy helper method. Do not use it for new tests. This is a legacy helper method. Do not use it for new tests.
@ -58,7 +58,7 @@ def train_group(hass, group):
hass.async_create_task(hass.services.async_call(DOMAIN, SERVICE_TRAIN_GROUP, data)) hass.async_create_task(hass.services.async_call(DOMAIN, SERVICE_TRAIN_GROUP, data))
def create_person(hass, group, name): def create_person(hass: HomeAssistant, group: str, name: str) -> None:
"""Create a person in a group. """Create a person in a group.
This is a legacy helper method. Do not use it for new tests. This is a legacy helper method. Do not use it for new tests.
@ -69,7 +69,7 @@ def create_person(hass, group, name):
) )
def delete_person(hass, group, name): def delete_person(hass: HomeAssistant, group: str, name: str) -> None:
"""Delete a person in a group. """Delete a person in a group.
This is a legacy helper method. Do not use it for new tests. This is a legacy helper method. Do not use it for new tests.
@ -80,7 +80,9 @@ def delete_person(hass, group, name):
) )
def face_person(hass, group, person, camera_entity): def face_person(
hass: HomeAssistant, group: str, person: str, camera_entity: str
) -> None:
"""Add a new face picture to a person. """Add a new face picture to a person.
This is a legacy helper method. Do not use it for new tests. This is a legacy helper method. Do not use it for new tests.

View File

@ -1,6 +1,7 @@
"""The tests for Monoprice Media player platform.""" """The tests for Monoprice Media player platform."""
from collections import defaultdict from collections import defaultdict
from typing import Any
from unittest.mock import patch from unittest.mock import patch
from serial import SerialException from serial import SerialException
@ -105,7 +106,7 @@ async def test_cannot_connect(hass: HomeAssistant) -> None:
assert hass.states.get(ZONE_1_ID) is None assert hass.states.get(ZONE_1_ID) is None
async def _setup_monoprice(hass, monoprice): async def _setup_monoprice(hass: HomeAssistant, monoprice: MockMonoprice) -> None:
with patch( with patch(
"homeassistant.components.monoprice.get_monoprice", "homeassistant.components.monoprice.get_monoprice",
new=lambda *a: monoprice, new=lambda *a: monoprice,
@ -116,7 +117,9 @@ async def _setup_monoprice(hass, monoprice):
await hass.async_block_till_done() await hass.async_block_till_done()
async def _setup_monoprice_with_options(hass, monoprice): async def _setup_monoprice_with_options(
hass: HomeAssistant, monoprice: MockMonoprice
) -> None:
with patch( with patch(
"homeassistant.components.monoprice.get_monoprice", "homeassistant.components.monoprice.get_monoprice",
new=lambda *a: monoprice, new=lambda *a: monoprice,
@ -129,7 +132,9 @@ async def _setup_monoprice_with_options(hass, monoprice):
await hass.async_block_till_done() await hass.async_block_till_done()
async def _setup_monoprice_not_first_run(hass, monoprice): async def _setup_monoprice_not_first_run(
hass: HomeAssistant, monoprice: MockMonoprice
) -> None:
with patch( with patch(
"homeassistant.components.monoprice.get_monoprice", "homeassistant.components.monoprice.get_monoprice",
new=lambda *a: monoprice, new=lambda *a: monoprice,
@ -141,19 +146,17 @@ async def _setup_monoprice_not_first_run(hass, monoprice):
await hass.async_block_till_done() await hass.async_block_till_done()
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( await hass.services.async_call(
MEDIA_PLAYER_DOMAIN, name, service_data=data, blocking=True MEDIA_PLAYER_DOMAIN, name, service_data=data, blocking=True
) )
async def _call_homeassistant_service(hass, name, data): async def _call_monoprice_service(
await hass.services.async_call( hass: HomeAssistant, name: str, data: dict[str, Any]
"homeassistant", name, service_data=data, blocking=True ) -> None:
)
async def _call_monoprice_service(hass, name, data):
await hass.services.async_call(DOMAIN, name, service_data=data, blocking=True) await hass.services.async_call(DOMAIN, name, service_data=data, blocking=True)

View File

@ -20,7 +20,12 @@ from tests.common import (
from tests.typing import MqttMockHAClient from tests.typing import MqttMockHAClient
async def add_eventstream(hass, sub_topic=None, pub_topic=None, ignore_event=None): async def add_eventstream(
hass: HomeAssistant,
sub_topic: str | None = None,
pub_topic: str | None = None,
ignore_event: list[str] | None = None,
) -> bool:
"""Add a mqtt_eventstream component.""" """Add a mqtt_eventstream component."""
config = {} config = {}
if sub_topic: if sub_topic:

View File

@ -11,11 +11,13 @@ import pytest
from homeassistant.components.device_tracker.legacy import ( from homeassistant.components.device_tracker.legacy import (
DOMAIN as DT_DOMAIN, DOMAIN as DT_DOMAIN,
YAML_DEVICES, YAML_DEVICES,
AsyncSeeCallback,
) )
from homeassistant.components.mqtt import DOMAIN as MQTT_DOMAIN from homeassistant.components.mqtt import DOMAIN as MQTT_DOMAIN
from homeassistant.config_entries import ConfigEntryDisabler from homeassistant.config_entries import ConfigEntryDisabler
from homeassistant.const import CONF_PLATFORM from homeassistant.const import CONF_PLATFORM
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from tests.common import async_fire_mqtt_message from tests.common import async_fire_mqtt_message
@ -71,9 +73,15 @@ async def test_setup_fails_without_mqtt_being_setup(
async def test_ensure_device_tracker_platform_validation(hass: HomeAssistant) -> None: async def test_ensure_device_tracker_platform_validation(hass: HomeAssistant) -> None:
"""Test if platform validation was done.""" """Test if platform validation was done."""
async def mock_setup_scanner(hass, config, see, discovery_info=None): async def mock_setup_scanner(
hass: HomeAssistant,
config: ConfigType,
see: AsyncSeeCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> bool:
"""Check that Qos was added by validation.""" """Check that Qos was added by validation."""
assert "qos" in config assert "qos" in config
return True
with patch( with patch(
"homeassistant.components.mqtt_json.device_tracker.async_setup_scanner", "homeassistant.components.mqtt_json.device_tracker.async_setup_scanner",

View File

@ -2,6 +2,7 @@
import datetime import datetime
import json import json
from typing import Any
from unittest.mock import patch from unittest.mock import patch
import pytest import pytest
@ -40,20 +41,22 @@ FAR_MESSAGE = {"id": DEVICE_ID, "name": NAME, "distance": 10}
REALLY_FAR_MESSAGE = {"id": DEVICE_ID, "name": NAME, "distance": 20} REALLY_FAR_MESSAGE = {"id": DEVICE_ID, "name": NAME, "distance": 20}
async def send_message(hass, topic, message): async def send_message(
hass: HomeAssistant, topic: str, message: dict[str, Any]
) -> None:
"""Test the sending of a message.""" """Test the sending of a message."""
async_fire_mqtt_message(hass, topic, json.dumps(message)) async_fire_mqtt_message(hass, topic, json.dumps(message))
await hass.async_block_till_done() await hass.async_block_till_done()
await hass.async_block_till_done() await hass.async_block_till_done()
async def assert_state(hass, room): async def assert_state(hass: HomeAssistant, room: str) -> None:
"""Test the assertion of a room state.""" """Test the assertion of a room state."""
state = hass.states.get(SENSOR_STATE) state = hass.states.get(SENSOR_STATE)
assert state.state == room assert state.state == room
async def assert_distance(hass, distance): async def assert_distance(hass: HomeAssistant, distance: int) -> None:
"""Test the assertion of a distance state.""" """Test the assertion of a distance state."""
state = hass.states.get(SENSOR_STATE) state = hass.states.get(SENSOR_STATE)
assert state.attributes.get("distance") == distance assert state.attributes.get("distance") == distance