jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-09 22:37:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

Correct type hints on MQTT tests (#128299)

Browse Source
This commit is contained in:
Jan Bouwhuis 2024-10-16 13:49:18 +02:00 committed by GitHub
parent 9f2bdca9ad
commit ac6d893758
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
23 changed files with 127 additions and 120 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tests/components/light/common.py
View File

@ -99,7 +99,7 @@ async def async_turn_on(
flash: str | None = None,
effect: str | None = None,
color_name: str | None = None,
white: bool | None = None,
white: int | None = None,
) -> None:
"""Turn all or specified light on."""
data = {

2
tests/components/mqtt/test_binary_sensor.py
View File

@ -1133,7 +1133,7 @@ async def test_skip_restoring_state_with_over_due_expire_trigger(
freezer.move_to("2022-02-02 12:02:00+01:00")
domain = binary_sensor.DOMAIN
config3 = copy.deepcopy(DEFAULT_CONFIG[mqtt.DOMAIN][domain])
config3: ConfigType = copy.deepcopy(DEFAULT_CONFIG[mqtt.DOMAIN][domain])
config3["name"] = "test3"
config3["expire_after"] = 10
config3["state_topic"] = "test-topic3"

12
tests/components/mqtt/test_client.py
View File

@ -1,9 +1,10 @@
"""The tests for the MQTT client."""
import asyncio
from datetime import datetime, timedelta
from datetime import timedelta
import socket
import ssl
import time
from typing import Any
from unittest.mock import MagicMock, Mock, call, patch
@ -296,10 +297,13 @@ async def test_subscribe_mqtt_config_entry_disabled(
mqtt_mock.connected = True
mqtt_config_entry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
assert mqtt_config_entry.state is ConfigEntryState.LOADED
mqtt_config_entry_state = mqtt_config_entry.state
assert mqtt_config_entry_state is ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(mqtt_config_entry.entry_id)
assert mqtt_config_entry.state is ConfigEntryState.NOT_LOADED
mqtt_config_entry_state = mqtt_config_entry.state
assert mqtt_config_entry_state is ConfigEntryState.NOT_LOADED
await hass.config_entries.async_set_disabled_by(
mqtt_config_entry.entry_id, ConfigEntryDisabler.USER
@ -1279,7 +1283,7 @@ async def test_handle_message_callback(
callbacks.append(args)
msg = ReceiveMessage(
"some-topic", b"test-payload", 1, False, "some-topic", datetime.now()
"some-topic", b"test-payload", 1, False, "some-topic", time.monotonic()
)
mock_debouncer.clear()
await mqtt.async_subscribe(hass, "some-topic", _callback)

35
tests/components/mqtt/test_climate.py
View File

@ -202,7 +202,7 @@ async def test_set_operation_bad_attr_and_state(
state = hass.states.get(ENTITY_CLIMATE)
assert state.state == "off"
with pytest.raises(vol.Invalid) as excinfo:
await common.async_set_hvac_mode(hass, None, ENTITY_CLIMATE)
await common.async_set_hvac_mode(hass, None, ENTITY_CLIMATE) # type:ignore[arg-type]
assert (
"expected HVACMode or one of 'off', 'heat', 'cool', 'heat_cool', 'auto', 'dry',"
" 'fan_only' for dictionary value @ data['hvac_mode']" in str(excinfo.value)
@ -220,10 +220,9 @@ async def test_set_operation(
state = hass.states.get(ENTITY_CLIMATE)
assert state.state == "off"
await common.async_set_hvac_mode(hass, "cool", ENTITY_CLIMATE)
await common.async_set_hvac_mode(hass, HVACMode.COOL, ENTITY_CLIMATE)
state = hass.states.get(ENTITY_CLIMATE)
assert state.state == "cool"
assert state.state == "cool"
mqtt_mock.async_publish.assert_called_once_with("mode-topic", "cool", 0, False)
@ -245,7 +244,7 @@ async def test_set_operation_pessimistic(
state = hass.states.get(ENTITY_CLIMATE)
assert state.state == STATE_UNKNOWN
await common.async_set_hvac_mode(hass, "cool", ENTITY_CLIMATE)
await common.async_set_hvac_mode(hass, HVACMode.COOL, ENTITY_CLIMATE)
state = hass.states.get(ENTITY_CLIMATE)
assert state.state == STATE_UNKNOWN
@ -287,7 +286,7 @@ async def test_set_operation_optimistic(
state = hass.states.get(ENTITY_CLIMATE)
assert state.state == "off"
await common.async_set_hvac_mode(hass, "cool", ENTITY_CLIMATE)
await common.async_set_hvac_mode(hass, HVACMode.COOL, ENTITY_CLIMATE)
state = hass.states.get(ENTITY_CLIMATE)
assert state.state == "cool"
@ -316,13 +315,13 @@ async def test_set_operation_with_power_command(
state = hass.states.get(ENTITY_CLIMATE)
assert state.state == "off"
await common.async_set_hvac_mode(hass, "cool", ENTITY_CLIMATE)
await common.async_set_hvac_mode(hass, HVACMode.COOL, ENTITY_CLIMATE)
state = hass.states.get(ENTITY_CLIMATE)
assert state.state == "cool"
mqtt_mock.async_publish.assert_has_calls([call("mode-topic", "cool", 0, False)])
mqtt_mock.async_publish.reset_mock()
await common.async_set_hvac_mode(hass, "off", ENTITY_CLIMATE)
await common.async_set_hvac_mode(hass, HVACMode.OFF, ENTITY_CLIMATE)
state = hass.states.get(ENTITY_CLIMATE)
assert state.state == "off"
mqtt_mock.async_publish.assert_has_calls([call("mode-topic", "off", 0, False)])
@ -358,12 +357,12 @@ async def test_turn_on_and_off_optimistic_with_power_command(
state = hass.states.get(ENTITY_CLIMATE)
assert state.state == "off"
await common.async_set_hvac_mode(hass, "cool", ENTITY_CLIMATE)
await common.async_set_hvac_mode(hass, HVACMode.COOL, ENTITY_CLIMATE)
state = hass.states.get(ENTITY_CLIMATE)
assert state.state == "cool"
mqtt_mock.async_publish.assert_has_calls([call("mode-topic", "cool", 0, False)])
mqtt_mock.async_publish.reset_mock()
await common.async_set_hvac_mode(hass, "off", ENTITY_CLIMATE)
await common.async_set_hvac_mode(hass, HVACMode.OFF, ENTITY_CLIMATE)
state = hass.states.get(ENTITY_CLIMATE)
assert state.state == "off"
@ -374,7 +373,7 @@ async def test_turn_on_and_off_optimistic_with_power_command(
mqtt_mock.async_publish.assert_has_calls([call("power-command", "ON", 0, False)])
mqtt_mock.async_publish.reset_mock()
await common.async_set_hvac_mode(hass, "cool", ENTITY_CLIMATE)
await common.async_set_hvac_mode(hass, HVACMode.COOL, ENTITY_CLIMATE)
state = hass.states.get(ENTITY_CLIMATE)
assert state.state == "cool"
await common.async_turn_off(hass, ENTITY_CLIMATE)
@ -433,7 +432,7 @@ async def test_turn_on_and_off_without_power_command(
else:
mqtt_mock.async_publish.assert_has_calls([])
await common.async_set_hvac_mode(hass, "cool", ENTITY_CLIMATE)
await common.async_set_hvac_mode(hass, HVACMode.COOL, ENTITY_CLIMATE)
state = hass.states.get(ENTITY_CLIMATE)
assert state.state == "cool"
mqtt_mock.async_publish.reset_mock()
@ -460,7 +459,7 @@ async def test_set_fan_mode_bad_attr(
state = hass.states.get(ENTITY_CLIMATE)
assert state.attributes.get("fan_mode") == "low"
with pytest.raises(vol.Invalid) as excinfo:
await common.async_set_fan_mode(hass, None, ENTITY_CLIMATE)
await common.async_set_fan_mode(hass, None, ENTITY_CLIMATE) # type:ignore[arg-type]
assert "string value is None for dictionary value @ data['fan_mode']" in str(
excinfo.value
)
@ -555,7 +554,7 @@ async def test_set_swing_mode_bad_attr(
state = hass.states.get(ENTITY_CLIMATE)
assert state.attributes.get("swing_mode") == "off"
with pytest.raises(vol.Invalid) as excinfo:
await common.async_set_swing_mode(hass, None, ENTITY_CLIMATE)
await common.async_set_swing_mode(hass, None, ENTITY_CLIMATE) # type:ignore[arg-type]
assert "string value is None for dictionary value @ data['swing_mode']" in str(
excinfo.value
)
@ -649,7 +648,7 @@ async def test_set_target_temperature(
state = hass.states.get(ENTITY_CLIMATE)
assert state.attributes.get("temperature") == 21
await common.async_set_hvac_mode(hass, "heat", ENTITY_CLIMATE)
await common.async_set_hvac_mode(hass, HVACMode.HEAT, ENTITY_CLIMATE)
state = hass.states.get(ENTITY_CLIMATE)
assert state.state == "heat"
mqtt_mock.async_publish.assert_called_once_with("mode-topic", "heat", 0, False)
@ -712,7 +711,7 @@ async def test_set_target_temperature_pessimistic(
state = hass.states.get(ENTITY_CLIMATE)
assert state.attributes.get("temperature") is None
await common.async_set_hvac_mode(hass, "heat", ENTITY_CLIMATE)
await common.async_set_hvac_mode(hass, HVACMode.HEAT, ENTITY_CLIMATE)
await common.async_set_temperature(hass, temperature=35, entity_id=ENTITY_CLIMATE)
state = hass.states.get(ENTITY_CLIMATE)
assert state.attributes.get("temperature") is None
@ -744,7 +743,7 @@ async def test_set_target_temperature_optimistic(
state = hass.states.get(ENTITY_CLIMATE)
assert state.attributes.get("temperature") == 21
await common.async_set_hvac_mode(hass, "heat", ENTITY_CLIMATE)
await common.async_set_hvac_mode(hass, HVACMode.HEAT, ENTITY_CLIMATE)
await common.async_set_temperature(hass, temperature=17, entity_id=ENTITY_CLIMATE)
state = hass.states.get(ENTITY_CLIMATE)
assert state.attributes.get("temperature") == 17
@ -1547,14 +1546,14 @@ async def test_set_and_templates(
assert state.attributes.get("preset_mode") == PRESET_ECO
# Mode
await common.async_set_hvac_mode(hass, "cool", ENTITY_CLIMATE)
await common.async_set_hvac_mode(hass, HVACMode.COOL, ENTITY_CLIMATE)
mqtt_mock.async_publish.assert_any_call("mode-topic", "mode: cool", 0, False)
assert mqtt_mock.async_publish.call_count == 1
mqtt_mock.async_publish.reset_mock()
state = hass.states.get(ENTITY_CLIMATE)
assert state.state == "cool"
await common.async_set_hvac_mode(hass, "off", ENTITY_CLIMATE)
await common.async_set_hvac_mode(hass, HVACMode.OFF, ENTITY_CLIMATE)
mqtt_mock.async_publish.assert_any_call("mode-topic", "mode: off", 0, False)
assert mqtt_mock.async_publish.call_count == 1
mqtt_mock.async_publish.reset_mock()

11
tests/components/mqtt/test_common.py
View File

@ -72,7 +72,10 @@ DISCOVERY_COUNT = len(MQTT)
type _MqttMessageType = list[tuple[str, str]]
type _AttributesType = list[tuple[str, Any]]
type _StateDataType = list[tuple[_MqttMessageType, str | None, _AttributesType | None]]
type _StateDataType = (
list[tuple[_MqttMessageType, str, _AttributesType | None]]
| list[tuple[_MqttMessageType, str, None]]
)
def help_all_subscribe_calls(mqtt_client_mock: MqttMockPahoClient) -> list[Any]:
@ -106,7 +109,7 @@ def help_custom_config(
)
base.update(instance)
entity_instances.append(base)
config[mqtt.DOMAIN][mqtt_entity_domain]: list[ConfigType] = entity_instances
config[mqtt.DOMAIN][mqtt_entity_domain] = entity_instances
return config
@ -1360,11 +1363,11 @@ async def help_test_entity_debug_info_message(
mqtt_mock_entry: MqttMockHAClientGenerator,
domain: str,
config: ConfigType,
service: str,
service: str | None,
command_topic: str | None = None,
command_payload: str | None = None,
state_topic: str | object | None = _SENTINEL,
state_payload: str | None = None,
state_payload: bytes | str | None = None,
service_parameters: dict[str, Any] | None = None,
) -> None:
"""Test debug_info.

7
tests/components/mqtt/test_config_flow.py
View File

@ -455,8 +455,6 @@ async def test_hassio_confirm(
mock_finish_setup: MagicMock,
) -> None:
"""Test we can finish a config flow."""
mock_try_connection.return_value = True
result = await hass.config_entries.flow.async_init(
"mqtt",
data=HassioServiceInfo(
@ -1027,7 +1025,6 @@ async def test_bad_certificate(
test_input.pop(mqtt.CONF_CLIENT_KEY)
mqtt_mock = await mqtt_mock_entry()
mock_try_connection.return_value = True
config_entry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
# Add at least one advanced option to get the full form
hass.config_entries.async_update_entry(
@ -1276,7 +1273,7 @@ async def test_invalid_discovery_prefix(
def get_default(schema: vol.Schema, key: str) -> Any | None:
"""Get default value for key in voluptuous schema."""
for schema_key in schema:
for schema_key in schema: # type:ignore[attr-defined]
if schema_key == key:
if schema_key.default == vol.UNDEFINED:
return None
@ -1286,7 +1283,7 @@ def get_default(schema: vol.Schema, key: str) -> Any | None:
def get_suggested(schema: vol.Schema, key: str) -> Any | None:
"""Get suggested value for key in voluptuous schema."""
for schema_key in schema:
for schema_key in schema: # type:ignore[attr-defined]
if schema_key == key:
if (
schema_key.description is None

8
tests/components/mqtt/test_device_trigger.py
View File

@ -45,7 +45,7 @@ async def test_get_triggers(
await hass.async_block_till_done()
device_entry = device_registry.async_get_device(identifiers={("mqtt", "0AFFD2")})
expected_triggers = [
expected_triggers: list[dict[str, Any]] = [
{
"platform": "device",
"domain": DOMAIN,
@ -165,7 +165,7 @@ async def test_discover_bad_triggers(
await hass.async_block_till_done()
device_entry = device_registry.async_get_device(identifiers={("mqtt", "0AFFD2")})
expected_triggers = [
expected_triggers: list[dict[str, Any]] = [
{
"platform": "device",
"domain": DOMAIN,
@ -226,7 +226,7 @@ async def test_update_remove_triggers(
device_entry = device_registry.async_get_device(identifiers={("mqtt", "0AFFD2")})
assert device_entry.name == "milk"
expected_triggers1 = [
expected_triggers1: list[dict[str, Any]] = [
{
"platform": "device",
"domain": DOMAIN,
@ -1263,7 +1263,7 @@ async def test_entity_device_info_update(
"""Test device registry update."""
await mqtt_mock_entry()
config = {
config: dict[str, Any] = {
"automation_type": "trigger",
"topic": "test-topic",
"type": "foo",

4
tests/components/mqtt/test_fan.py
View File

@ -1486,7 +1486,7 @@ async def test_encoding_subscribable_topics(
attribute_value: Any,
) -> None:
"""Test handling of incoming encoded payload."""
config = copy.deepcopy(DEFAULT_CONFIG[mqtt.DOMAIN][fan.DOMAIN])
config: dict[str, Any] = copy.deepcopy(DEFAULT_CONFIG[mqtt.DOMAIN][fan.DOMAIN])
config[ATTR_PRESET_MODES] = ["eco", "auto"]
config[CONF_PRESET_MODE_COMMAND_TOPIC] = "fan/some_preset_mode_command_topic"
config[CONF_PERCENTAGE_COMMAND_TOPIC] = "fan/some_percentage_command_topic"
@ -2201,7 +2201,7 @@ async def test_publishing_with_custom_encoding(
) -> None:
"""Test publishing MQTT payload with different encoding."""
domain = fan.DOMAIN
config = copy.deepcopy(DEFAULT_CONFIG)
config: dict[str, Any] = copy.deepcopy(DEFAULT_CONFIG)
if topic == "preset_mode_command_topic":
config[mqtt.DOMAIN][domain]["preset_modes"] = ["auto", "eco"]

6
tests/components/mqtt/test_humidifier.py
View File

@ -862,7 +862,9 @@ async def test_encoding_subscribable_topics(
attribute_value: Any,
) -> None:
"""Test handling of incoming encoded payload."""
config = copy.deepcopy(DEFAULT_CONFIG[mqtt.DOMAIN][humidifier.DOMAIN])
config: dict[str, Any] = copy.deepcopy(
DEFAULT_CONFIG[mqtt.DOMAIN][humidifier.DOMAIN]
)
config["modes"] = ["eco", "auto"]
config[CONF_MODE_COMMAND_TOPIC] = "humidifier/some_mode_command_topic"
await help_test_encoding_subscribable_topics(
@ -1473,7 +1475,7 @@ async def test_publishing_with_custom_encoding(
) -> None:
"""Test publishing MQTT payload with different encoding."""
domain = humidifier.DOMAIN
config = copy.deepcopy(DEFAULT_CONFIG)
config: dict[str, Any] = copy.deepcopy(DEFAULT_CONFIG)
if topic == "mode_command_topic":
config[mqtt.DOMAIN][domain]["modes"] = ["auto", "eco"]

6
tests/components/mqtt/test_init.py
View File

@ -230,7 +230,7 @@ async def test_value_template_fails(hass: HomeAssistant) -> None:
)
with pytest.raises(MqttValueTemplateException) as exc:
val_tpl.async_render_with_possible_json_value(
'{"some_var": null }', default=100
'{"some_var": null }', default="100"
)
assert str(exc.value) == (
"TypeError: unsupported operand type(s) for *: 'NoneType' and 'int' "
@ -835,7 +835,7 @@ async def test_receiving_message_with_non_utf8_topic_gets_logged(
msg.payload = b"Payload"
msg.qos = 2
msg.retain = True
msg.timestamp = time.monotonic()
msg.timestamp = time.monotonic() # type:ignore[assignment]
mqtt_data: MqttData = hass.data["mqtt"]
assert mqtt_data.client
@ -1489,7 +1489,7 @@ async def test_debug_info_non_mqtt(
"""Test we get empty debug_info for a device with non MQTT entities."""
await mqtt_mock_entry()
domain = "sensor"
setup_test_component_platform(hass, domain, mock_sensor_entities)
setup_test_component_platform(hass, domain, mock_sensor_entities.values())
config_entry = MockConfigEntry(domain="test", data={})
config_entry.add_to_hass(hass)

4
tests/components/mqtt/test_lawn_mower.py
View File

@ -802,7 +802,9 @@ async def test_encoding_subscribable_topics(
attribute_value: Any,
) -> None:
"""Test handling of incoming encoded payload."""
config = copy.deepcopy(DEFAULT_CONFIG[mqtt.DOMAIN][lawn_mower.DOMAIN])
config: dict[str, Any] = copy.deepcopy(
DEFAULT_CONFIG[mqtt.DOMAIN][lawn_mower.DOMAIN]
)
config["actions"] = ["milk", "beer"]
await help_test_encoding_subscribable_topics(
hass,

30
tests/components/mqtt/test_light.py
View File

@ -1053,7 +1053,7 @@ async def test_sending_mqtt_commands_and_optimistic(
assert state.attributes.get(light.ATTR_SUPPORTED_COLOR_MODES) == color_modes
await common.async_turn_on(
hass, "light.test", brightness=10, rgb_color=[80, 40, 20]
hass, "light.test", brightness=10, rgb_color=(80, 40, 20)
)
mqtt_mock.async_publish.assert_has_calls(
[
@ -1073,7 +1073,7 @@ async def test_sending_mqtt_commands_and_optimistic(
assert state.attributes.get(light.ATTR_SUPPORTED_COLOR_MODES) == color_modes
await common.async_turn_on(
hass, "light.test", brightness=20, rgbw_color=[80, 40, 20, 10]
hass, "light.test", brightness=20, rgbw_color=(80, 40, 20, 10)
)
mqtt_mock.async_publish.assert_has_calls(
[
@ -1093,7 +1093,7 @@ async def test_sending_mqtt_commands_and_optimistic(
assert state.attributes.get(light.ATTR_SUPPORTED_COLOR_MODES) == color_modes
await common.async_turn_on(
hass, "light.test", brightness=40, rgbww_color=[80, 40, 20, 10, 8]
hass, "light.test", brightness=40, rgbww_color=(80, 40, 20, 10, 8)
)
mqtt_mock.async_publish.assert_has_calls(
[
@ -1112,7 +1112,7 @@ async def test_sending_mqtt_commands_and_optimistic(
assert state.attributes.get(light.ATTR_COLOR_MODE) == "rgbww"
assert state.attributes.get(light.ATTR_SUPPORTED_COLOR_MODES) == color_modes
await common.async_turn_on(hass, "light.test", brightness=50, hs_color=[359, 78])
await common.async_turn_on(hass, "light.test", brightness=50, hs_color=(359, 78))
mqtt_mock.async_publish.assert_has_calls(
[
call("test_light_rgb/set", "on", 2, False),
@ -1130,7 +1130,7 @@ async def test_sending_mqtt_commands_and_optimistic(
assert state.attributes.get(light.ATTR_COLOR_MODE) == "hs"
assert state.attributes.get(light.ATTR_SUPPORTED_COLOR_MODES) == color_modes
await common.async_turn_on(hass, "light.test", brightness=60, xy_color=[0.2, 0.3])
await common.async_turn_on(hass, "light.test", brightness=60, xy_color=(0.2, 0.3))
mqtt_mock.async_publish.assert_has_calls(
[
call("test_light_rgb/set", "on", 2, False),
@ -1193,7 +1193,7 @@ async def test_sending_mqtt_rgb_command_with_template(
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
await common.async_turn_on(hass, "light.test", rgb_color=[255, 128, 64])
await common.async_turn_on(hass, "light.test", rgb_color=(255, 128, 64))
mqtt_mock.async_publish.assert_has_calls(
[
@ -1236,7 +1236,7 @@ async def test_sending_mqtt_rgbw_command_with_template(
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
await common.async_turn_on(hass, "light.test", rgbw_color=[255, 128, 64, 32])
await common.async_turn_on(hass, "light.test", rgbw_color=(255, 128, 64, 32))
mqtt_mock.async_publish.assert_has_calls(
[
@ -1279,7 +1279,7 @@ async def test_sending_mqtt_rgbww_command_with_template(
state = hass.states.get("light.test")
assert state.state == STATE_UNKNOWN
await common.async_turn_on(hass, "light.test", rgbww_color=[255, 128, 64, 32, 16])
await common.async_turn_on(hass, "light.test", rgbww_color=(255, 128, 64, 32, 16))
mqtt_mock.async_publish.assert_has_calls(
[
@ -1469,7 +1469,7 @@ async def test_on_command_brightness(
# Turn on w/ just a color to ensure brightness gets
# added and sent.
await common.async_turn_on(hass, "light.test", rgb_color=[255, 128, 0])
await common.async_turn_on(hass, "light.test", rgb_color=(255, 128, 0))
mqtt_mock.async_publish.assert_has_calls(
[
@ -1545,7 +1545,7 @@ async def test_on_command_brightness_scaled(
# Turn on w/ just a color to ensure brightness gets
# added and sent.
await common.async_turn_on(hass, "light.test", rgb_color=[255, 128, 0])
await common.async_turn_on(hass, "light.test", rgb_color=(255, 128, 0))
mqtt_mock.async_publish.assert_has_calls(
[
@ -1626,7 +1626,7 @@ async def test_on_command_rgb(
mqtt_mock.async_publish.assert_called_once_with("test_light/set", "OFF", 0, False)
# Ensure color gets scaled with brightness.
await common.async_turn_on(hass, "light.test", rgb_color=[255, 128, 0])
await common.async_turn_on(hass, "light.test", rgb_color=(255, 128, 0))
mqtt_mock.async_publish.assert_has_calls(
[
@ -1722,7 +1722,7 @@ async def test_on_command_rgbw(
mqtt_mock.async_publish.assert_called_once_with("test_light/set", "OFF", 0, False)
# Ensure color gets scaled with brightness.
await common.async_turn_on(hass, "light.test", rgbw_color=[255, 128, 0, 16])
await common.async_turn_on(hass, "light.test", rgbw_color=(255, 128, 0, 16))
mqtt_mock.async_publish.assert_has_calls(
[
@ -1818,7 +1818,7 @@ async def test_on_command_rgbww(
mqtt_mock.async_publish.assert_called_once_with("test_light/set", "OFF", 0, False)
# Ensure color gets scaled with brightness.
await common.async_turn_on(hass, "light.test", rgbww_color=[255, 128, 0, 16, 32])
await common.async_turn_on(hass, "light.test", rgbww_color=(255, 128, 0, 16, 32))
mqtt_mock.async_publish.assert_has_calls(
[
@ -3262,7 +3262,7 @@ async def test_publishing_with_custom_encoding(
) -> None:
"""Test publishing MQTT payload with different encoding."""
domain = light.DOMAIN
config = copy.deepcopy(DEFAULT_CONFIG)
config: dict[str, Any] = copy.deepcopy(DEFAULT_CONFIG)
if topic == "effect_command_topic":
config[mqtt.DOMAIN][domain]["effect_list"] = ["random", "color_loop"]
elif topic == "white_command_topic":
@ -3333,7 +3333,7 @@ async def test_encoding_subscribable_topics(
init_payload: tuple[str, str] | None,
) -> None:
"""Test handling of incoming encoded payload."""
config = copy.deepcopy(DEFAULT_CONFIG[mqtt.DOMAIN][light.DOMAIN])
config: dict[str, Any] = copy.deepcopy(DEFAULT_CONFIG[mqtt.DOMAIN][light.DOMAIN])
config[CONF_EFFECT_COMMAND_TOPIC] = "light/CONF_EFFECT_COMMAND_TOPIC"
config[CONF_RGB_COMMAND_TOPIC] = "light/CONF_RGB_COMMAND_TOPIC"
config[CONF_BRIGHTNESS_COMMAND_TOPIC] = "light/CONF_BRIGHTNESS_COMMAND_TOPIC"

66
tests/components/mqtt/test_light_json.py
View File

@ -99,7 +99,7 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers.json import json_dumps
from homeassistant.util.json import JsonValueType, json_loads
from homeassistant.util.json import json_loads
from .test_common import (
help_custom_config,
@ -172,11 +172,11 @@ COLOR_MODES_CONFIG = {
class JsonValidator:
"""Helper to compare JSON."""
def __init__(self, jsondata: JsonValueType) -> None:
def __init__(self, jsondata: bytes | str) -> None:
"""Initialize JSON validator."""
self.jsondata = jsondata
def __eq__(self, other: JsonValueType) -> bool:
def __eq__(self, other: bytes | str) -> bool: # type:ignore[override]
"""Compare JSON data."""
return json_loads(self.jsondata) == json_loads(other)
@ -1108,7 +1108,7 @@ async def test_sending_mqtt_commands_and_optimistic(
mqtt_mock.reset_mock()
await common.async_turn_on(
hass, "light.test", brightness=50, xy_color=[0.123, 0.123]
hass, "light.test", brightness=50, xy_color=(0.123, 0.123)
)
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set",
@ -1128,7 +1128,7 @@ async def test_sending_mqtt_commands_and_optimistic(
assert state.attributes["rgb_color"] == (0, 123, 255)
assert state.attributes["xy_color"] == (0.14, 0.131)
await common.async_turn_on(hass, "light.test", brightness=50, hs_color=[359, 78])
await common.async_turn_on(hass, "light.test", brightness=50, hs_color=(359, 78))
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set",
JsonValidator(
@ -1148,7 +1148,7 @@ async def test_sending_mqtt_commands_and_optimistic(
assert state.attributes["rgb_color"] == (255, 56, 59)
assert state.attributes["xy_color"] == (0.654, 0.301)
await common.async_turn_on(hass, "light.test", rgb_color=[255, 128, 0])
await common.async_turn_on(hass, "light.test", rgb_color=(255, 128, 0))
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set",
JsonValidator(
@ -1265,7 +1265,7 @@ async def test_sending_mqtt_commands_and_optimistic2(
assert state.state == STATE_OFF
# Set hs color
await common.async_turn_on(hass, "light.test", brightness=75, hs_color=[359, 78])
await common.async_turn_on(hass, "light.test", brightness=75, hs_color=(359, 78))
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes["brightness"] == 75
@ -1286,7 +1286,7 @@ async def test_sending_mqtt_commands_and_optimistic2(
mqtt_mock.async_publish.reset_mock()
# Set rgb color
await common.async_turn_on(hass, "light.test", rgb_color=[255, 128, 0])
await common.async_turn_on(hass, "light.test", rgb_color=(255, 128, 0))
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes["brightness"] == 75
@ -1305,7 +1305,7 @@ async def test_sending_mqtt_commands_and_optimistic2(
mqtt_mock.async_publish.reset_mock()
# Set rgbw color
await common.async_turn_on(hass, "light.test", rgbw_color=[255, 128, 0, 123])
await common.async_turn_on(hass, "light.test", rgbw_color=(255, 128, 0, 123))
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes["brightness"] == 75
@ -1326,7 +1326,7 @@ async def test_sending_mqtt_commands_and_optimistic2(
mqtt_mock.async_publish.reset_mock()
# Set rgbww color
await common.async_turn_on(hass, "light.test", rgbww_color=[255, 128, 0, 45, 32])
await common.async_turn_on(hass, "light.test", rgbww_color=(255, 128, 0, 45, 32))
state = hass.states.get("light.test")
assert state.state == STATE_ON
assert state.attributes["brightness"] == 75
@ -1348,7 +1348,7 @@ async def test_sending_mqtt_commands_and_optimistic2(
# Set xy color
await common.async_turn_on(
hass, "light.test", brightness=50, xy_color=[0.123, 0.223]
hass, "light.test", brightness=50, xy_color=(0.123, 0.223)
)
state = hass.states.get("light.test")
assert state.state == STATE_ON
@ -1435,10 +1435,10 @@ async def test_sending_hs_color(
mqtt_mock.reset_mock()
await common.async_turn_on(
hass, "light.test", brightness=50, xy_color=[0.123, 0.123]
hass, "light.test", brightness=50, xy_color=(0.123, 0.123)
)
await common.async_turn_on(hass, "light.test", brightness=50, hs_color=[359, 78])
await common.async_turn_on(hass, "light.test", rgb_color=[255, 128, 0])
await common.async_turn_on(hass, "light.test", brightness=50, hs_color=(359, 78))
await common.async_turn_on(hass, "light.test", rgb_color=(255, 128, 0))
mqtt_mock.async_publish.assert_has_calls(
[
@ -1497,11 +1497,11 @@ async def test_sending_rgb_color_no_brightness(
assert state.state == STATE_UNKNOWN
await common.async_turn_on(
hass, "light.test", brightness=50, xy_color=[0.123, 0.123]
hass, "light.test", brightness=50, xy_color=(0.123, 0.123)
)
await common.async_turn_on(hass, "light.test", brightness=50, hs_color=[359, 78])
await common.async_turn_on(hass, "light.test", brightness=50, hs_color=(359, 78))
await common.async_turn_on(
hass, "light.test", rgb_color=[255, 128, 0], brightness=255
hass, "light.test", rgb_color=(255, 128, 0), brightness=255
)
mqtt_mock.async_publish.assert_has_calls(
@ -1555,17 +1555,17 @@ async def test_sending_rgb_color_no_brightness2(
assert state.state == STATE_UNKNOWN
await common.async_turn_on(
hass, "light.test", brightness=50, xy_color=[0.123, 0.123]
hass, "light.test", brightness=50, xy_color=(0.123, 0.123)
)
await common.async_turn_on(hass, "light.test", brightness=50, hs_color=[359, 78])
await common.async_turn_on(hass, "light.test", brightness=50, hs_color=(359, 78))
await common.async_turn_on(
hass, "light.test", rgb_color=[255, 128, 0], brightness=255
hass, "light.test", rgb_color=(255, 128, 0), brightness=255
)
await common.async_turn_on(
hass, "light.test", rgbw_color=[128, 64, 32, 16], brightness=128
hass, "light.test", rgbw_color=(128, 64, 32, 16), brightness=128
)
await common.async_turn_on(
hass, "light.test", rgbww_color=[128, 64, 32, 16, 8], brightness=64
hass, "light.test", rgbww_color=(128, 64, 32, 16, 8), brightness=64
)
mqtt_mock.async_publish.assert_has_calls(
@ -1635,11 +1635,11 @@ async def test_sending_rgb_color_with_brightness(
assert state.state == STATE_UNKNOWN
await common.async_turn_on(
hass, "light.test", brightness=50, xy_color=[0.123, 0.123]
hass, "light.test", brightness=50, xy_color=(0.123, 0.123)
)
await common.async_turn_on(hass, "light.test", brightness=255, hs_color=[359, 78])
await common.async_turn_on(hass, "light.test", brightness=255, hs_color=(359, 78))
await common.async_turn_on(hass, "light.test", brightness=1)
await common.async_turn_on(hass, "light.test", rgb_color=[255, 128, 0])
await common.async_turn_on(hass, "light.test", rgb_color=(255, 128, 0))
mqtt_mock.async_publish.assert_has_calls(
[
@ -1705,11 +1705,11 @@ async def test_sending_rgb_color_with_scaled_brightness(
assert state.state == STATE_UNKNOWN
await common.async_turn_on(
hass, "light.test", brightness=50, xy_color=[0.123, 0.123]
hass, "light.test", brightness=50, xy_color=(0.123, 0.123)
)
await common.async_turn_on(hass, "light.test", brightness=255, hs_color=[359, 78])
await common.async_turn_on(hass, "light.test", brightness=255, hs_color=(359, 78))
await common.async_turn_on(hass, "light.test", brightness=1)
await common.async_turn_on(hass, "light.test", rgb_color=[255, 128, 0])
await common.async_turn_on(hass, "light.test", rgb_color=(255, 128, 0))
mqtt_mock.async_publish.assert_has_calls(
[
@ -1820,10 +1820,10 @@ async def test_sending_xy_color(
assert state.state == STATE_UNKNOWN
await common.async_turn_on(
hass, "light.test", brightness=50, xy_color=[0.123, 0.123]
hass, "light.test", brightness=50, xy_color=(0.123, 0.123)
)
await common.async_turn_on(hass, "light.test", brightness=50, hs_color=[359, 78])
await common.async_turn_on(hass, "light.test", rgb_color=[255, 128, 0])
await common.async_turn_on(hass, "light.test", brightness=50, hs_color=(359, 78))
await common.async_turn_on(hass, "light.test", rgb_color=(255, 128, 0))
mqtt_mock.async_publish.assert_has_calls(
[
@ -2629,7 +2629,7 @@ async def test_publishing_with_custom_encoding(
) -> None:
"""Test publishing MQTT payload with different encoding."""
domain = light.DOMAIN
config = copy.deepcopy(DEFAULT_CONFIG)
config: dict[str, Any] = copy.deepcopy(DEFAULT_CONFIG)
if topic == "effect_command_topic":
config[mqtt.DOMAIN][domain]["effect_list"] = ["random", "color_loop"]
@ -2680,7 +2680,7 @@ async def test_encoding_subscribable_topics(
init_payload: tuple[str, str] | None,
) -> None:
"""Test handling of incoming encoded payload."""
config = copy.deepcopy(DEFAULT_CONFIG[mqtt.DOMAIN][light.DOMAIN])
config: dict[str, Any] = copy.deepcopy(DEFAULT_CONFIG[mqtt.DOMAIN][light.DOMAIN])
config["color_mode"] = True
config["supported_color_modes"] = [
"color_temp",

18
tests/components/mqtt/test_light_template.py
View File

@ -482,7 +482,7 @@ async def test_sending_mqtt_commands_and_optimistic(
assert state.state == STATE_ON
# Full brightness - no scaling of RGB values sent over MQTT
await common.async_turn_on(hass, "light.test", rgb_color=[255, 128, 0])
await common.async_turn_on(hass, "light.test", rgb_color=(255, 128, 0))
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", "on,,,255-128-0,30.118-100.0", 2, False
)
@ -492,7 +492,7 @@ async def test_sending_mqtt_commands_and_optimistic(
assert state.attributes.get("rgb_color") == (255, 128, 0)
# Full brightness - normalization of RGB values sent over MQTT
await common.async_turn_on(hass, "light.test", rgb_color=[128, 64, 0])
await common.async_turn_on(hass, "light.test", rgb_color=(128, 64, 0))
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", "on,,,255-127-0,30.0-100.0", 2, False
)
@ -511,7 +511,7 @@ async def test_sending_mqtt_commands_and_optimistic(
assert state.state == STATE_ON
# Half brightness - scaling of RGB values sent over MQTT
await common.async_turn_on(hass, "light.test", rgb_color=[0, 255, 128])
await common.async_turn_on(hass, "light.test", rgb_color=(0, 255, 128))
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", "on,,,0-128-64,150.118-100.0", 2, False
)
@ -521,7 +521,7 @@ async def test_sending_mqtt_commands_and_optimistic(
assert state.attributes.get("rgb_color") == (0, 255, 128)
# Half brightness - normalization+scaling of RGB values sent over MQTT
await common.async_turn_on(hass, "light.test", rgb_color=[0, 32, 16])
await common.async_turn_on(hass, "light.test", rgb_color=(0, 32, 16))
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", "on,,,0-128-64,150.0-100.0", 2, False
)
@ -614,7 +614,7 @@ async def test_sending_mqtt_commands_non_optimistic_brightness_template(
assert not state.attributes.get("brightness")
# Full brightness - no scaling of RGB values sent over MQTT
await common.async_turn_on(hass, "light.test", rgb_color=[255, 128, 0])
await common.async_turn_on(hass, "light.test", rgb_color=(255, 128, 0))
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", "on,,,255-128-0,30.118-100.0", 0, False
)
@ -624,7 +624,7 @@ async def test_sending_mqtt_commands_non_optimistic_brightness_template(
assert not state.attributes.get("rgb_color")
# Full brightness - normalization of RGB values sent over MQTT
await common.async_turn_on(hass, "light.test", rgb_color=[128, 64, 0])
await common.async_turn_on(hass, "light.test", rgb_color=(128, 64, 0))
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", "on,,,255-127-0,30.0-100.0", 0, False
)
@ -638,7 +638,7 @@ async def test_sending_mqtt_commands_non_optimistic_brightness_template(
mqtt_mock.async_publish.reset_mock()
# Half brightness - no scaling of RGB values sent over MQTT
await common.async_turn_on(hass, "light.test", rgb_color=[0, 255, 128])
await common.async_turn_on(hass, "light.test", rgb_color=(0, 255, 128))
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", "on,,,0-255-128,150.118-100.0", 0, False
)
@ -646,7 +646,7 @@ async def test_sending_mqtt_commands_non_optimistic_brightness_template(
state = hass.states.get("light.test")
# Half brightness - normalization but no scaling of RGB values sent over MQTT
await common.async_turn_on(hass, "light.test", rgb_color=[0, 32, 16])
await common.async_turn_on(hass, "light.test", rgb_color=(0, 32, 16))
mqtt_mock.async_publish.assert_called_once_with(
"test_light_rgb/set", "on,,,0-255-127,150.0-100.0", 0, False
)
@ -1259,7 +1259,7 @@ async def test_publishing_with_custom_encoding(
) -> None:
"""Test publishing MQTT payload with different encoding."""
domain = light.DOMAIN
config = copy.deepcopy(DEFAULT_CONFIG)
config: dict[str, Any] = copy.deepcopy(DEFAULT_CONFIG)
if topic == "effect_command_topic":
config[mqtt.DOMAIN][domain]["effect_list"] = ["random", "color_loop"]

2
tests/components/mqtt/test_select.py
View File

@ -610,7 +610,7 @@ def _test_options_attributes_options_config(
@pytest.mark.parametrize(
("hass_config", "options"),
_test_options_attributes_options_config((["milk", "beer"], ["milk"], [])),
_test_options_attributes_options_config((["milk", "beer"], ["milk"], [])), # type:ignore[arg-type]
)
async def test_options_attributes(
hass: HomeAssistant, mqtt_mock_entry: MqttMockHAClientGenerator, options: list[str]

4
tests/components/mqtt/test_sensor.py
View File

@ -713,7 +713,7 @@ async def test_force_update_disabled(
def test_callback(event: Event) -> None:
events.append(event)
hass.bus.async_listen(EVENT_STATE_CHANGED, test_callback)
hass.bus.async_listen(EVENT_STATE_CHANGED, test_callback) # type:ignore[arg-type]
async_fire_mqtt_message(hass, "test-topic", "100")
await hass.async_block_till_done()
@ -751,7 +751,7 @@ async def test_force_update_enabled(
def test_callback(event: Event) -> None:
events.append(event)
hass.bus.async_listen(EVENT_STATE_CHANGED, test_callback)
hass.bus.async_listen(EVENT_STATE_CHANGED, test_callback) # type:ignore[arg-type]
async_fire_mqtt_message(hass, "test-topic", "100")
await hass.async_block_till_done()

4
tests/components/mqtt/test_siren.py
View File

@ -594,7 +594,7 @@ async def test_setting_blocked_attribute_via_mqtt_json_message(
) -> None:
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_blocked_attribute_via_mqtt_json_message(
hass, mqtt_mock_entry, siren.DOMAIN, DEFAULT_CONFIG, {}
hass, mqtt_mock_entry, siren.DOMAIN, DEFAULT_CONFIG, None
)
@ -974,7 +974,7 @@ async def test_publishing_with_custom_encoding(
) -> None:
"""Test publishing MQTT payload with command templates and different encoding."""
domain = siren.DOMAIN
config = copy.deepcopy(DEFAULT_CONFIG)
config: dict[str, Any] = copy.deepcopy(DEFAULT_CONFIG)
config[mqtt.DOMAIN][domain][siren.ATTR_AVAILABLE_TONES] = ["siren", "xylophone"]
await help_test_publishing_with_custom_encoding(

2
tests/components/mqtt/test_switch.py
View File

@ -403,7 +403,7 @@ async def test_setting_blocked_attribute_via_mqtt_json_message(
) -> None:
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_blocked_attribute_via_mqtt_json_message(
hass, mqtt_mock_entry, switch.DOMAIN, DEFAULT_CONFIG, {}
hass, mqtt_mock_entry, switch.DOMAIN, DEFAULT_CONFIG, None
)

3
tests/components/mqtt/test_tag.py
View File

@ -3,6 +3,7 @@
from collections.abc import Generator
import copy
import json
from typing import Any
from unittest.mock import ANY, AsyncMock, patch
import pytest
@ -504,7 +505,7 @@ async def test_entity_device_info_update(
"""Test device registry update."""
await mqtt_mock_entry()
config = {
config: dict[str, Any] = {
"topic": "test-topic",
"device": {
"identifiers": ["helloworld"],

2
tests/components/mqtt/test_text.py
View File

@ -469,7 +469,7 @@ async def test_setting_blocked_attribute_via_mqtt_json_message(
) -> None:
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_blocked_attribute_via_mqtt_json_message(
hass, mqtt_mock_entry, text.DOMAIN, DEFAULT_CONFIG, {}
hass, mqtt_mock_entry, text.DOMAIN, DEFAULT_CONFIG, None
)

15
tests/components/mqtt/test_util.py
View File

@ -236,8 +236,7 @@ async def test_waiting_for_client_not_loaded(
unsubs: list[Callable[[], None]] = []
async def _async_just_in_time_subscribe() -> Callable[[], None]:
nonlocal unsub
async def _async_just_in_time_subscribe() -> None:
assert await mqtt.async_wait_for_mqtt_client(hass)
# Awaiting a second time should work too and return True
assert await mqtt.async_wait_for_mqtt_client(hass)
@ -261,12 +260,12 @@ async def test_waiting_for_client_loaded(
"""Test waiting for client where mqtt entry is loaded."""
unsub: Callable[[], None] | None = None
async def _async_just_in_time_subscribe() -> Callable[[], None]:
async def _async_just_in_time_subscribe() -> None:
nonlocal unsub
assert await mqtt.async_wait_for_mqtt_client(hass)
unsub = await mqtt.async_subscribe(hass, "test_topic", lambda msg: None)
entry = hass.config_entries.async_entries(mqtt.DATA_MQTT)[0]
entry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
assert entry.state is ConfigEntryState.LOADED
await _async_just_in_time_subscribe()
@ -290,7 +289,7 @@ async def test_waiting_for_client_entry_fails(
)
entry.add_to_hass(hass)
async def _async_just_in_time_subscribe() -> Callable[[], None]:
async def _async_just_in_time_subscribe() -> None:
assert not await mqtt.async_wait_for_mqtt_client(hass)
hass.async_create_task(_async_just_in_time_subscribe())
@ -300,7 +299,7 @@ async def test_waiting_for_client_entry_fails(
side_effect=Exception,
):
await hass.config_entries.async_setup(entry.entry_id)
assert entry.state is ConfigEntryState.SETUP_ERROR
assert entry.state is ConfigEntryState.SETUP_ERROR # type:ignore[comparison-overlap]
async def test_waiting_for_client_setup_fails(
@ -318,7 +317,7 @@ async def test_waiting_for_client_setup_fails(
)
entry.add_to_hass(hass)
async def _async_just_in_time_subscribe() -> Callable[[], None]:
async def _async_just_in_time_subscribe() -> None:
assert not await mqtt.async_wait_for_mqtt_client(hass)
hass.async_create_task(_async_just_in_time_subscribe())
@ -327,7 +326,7 @@ async def test_waiting_for_client_setup_fails(
# Simulate MQTT setup fails before the client would become available
mqtt_client_mock.connect.side_effect = Exception
assert not await hass.config_entries.async_setup(entry.entry_id)
assert entry.state is ConfigEntryState.SETUP_ERROR
assert entry.state is ConfigEntryState.SETUP_ERROR # type:ignore[comparison-overlap]
@patch("homeassistant.components.mqtt.util.AVAILABILITY_TIMEOUT", 0.01)

2
tests/components/mqtt/test_vacuum.py
View File

@ -292,7 +292,7 @@ async def test_command_without_command_topic(
mqtt_mock.async_publish.assert_not_called()
mqtt_mock.async_publish.reset_mock()
await common.async_send_command(hass, "some command", "vacuum.test")
await common.async_send_command(hass, "some command", entity_id="vacuum.test")
mqtt_mock.async_publish.assert_not_called()
mqtt_mock.async_publish.reset_mock()

2
tests/components/mqtt/test_water_heater.py
View File

@ -162,7 +162,7 @@ async def test_set_operation_mode_bad_attr_and_state(
state = hass.states.get(ENTITY_WATER_HEATER)
assert state.state == "off"
with pytest.raises(vol.Invalid) as excinfo:
await common.async_set_operation_mode(hass, None, ENTITY_WATER_HEATER)
await common.async_set_operation_mode(hass, None, ENTITY_WATER_HEATER) # type:ignore[arg-type]
assert "string value is None for dictionary value @ data['operation_mode']" in str(
excinfo.value
)