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

Deduplicate MQTT tests (#32874)

Browse Source
This commit is contained in:
Erik Montnemery 2020-03-17 08:09:19 +01:00 committed by GitHub
parent a278cf3db2
commit 86d48c608e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 908 additions and 1747 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

231
tests/components/mqtt/common.py
View File

@ -1,9 +1,11 @@
"""Common test objects."""
import copy
import json
from unittest.mock import ANY
from homeassistant.components import mqtt
from homeassistant.components.mqtt.discovery import async_start
from homeassistant.const import ATTR_ASSUMED_STATE, STATE_UNAVAILABLE
from tests.common import (
MockConfigEntry,
@ -13,6 +15,123 @@ from tests.common import (
mock_registry,
)
DEFAULT_CONFIG_DEVICE_INFO_ID = {
"identifiers": ["helloworld"],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
}
DEFAULT_CONFIG_DEVICE_INFO_MAC = {
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
}
async def help_test_availability_without_topic(hass, mqtt_mock, domain, config):
"""Test availability without defined availability topic."""
assert "availability_topic" not in config[domain]
assert await async_setup_component(hass, domain, config)
state = hass.states.get(f"{domain}.test")
assert state.state != STATE_UNAVAILABLE
async def help_test_default_availability_payload(
hass,
mqtt_mock,
domain,
config,
no_assumed_state=False,
state_topic=None,
state_message=None,
):
"""Test availability by default payload with defined topic.
This is a test helper for the MqttAvailability mixin.
"""
# Add availability settings to config
config = copy.deepcopy(config)
config[domain]["availability_topic"] = "availability-topic"
assert await async_setup_component(hass, domain, config,)
state = hass.states.get(f"{domain}.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "online")
state = hass.states.get(f"{domain}.test")
assert state.state != STATE_UNAVAILABLE
if no_assumed_state:
assert not state.attributes.get(ATTR_ASSUMED_STATE)
async_fire_mqtt_message(hass, "availability-topic", "offline")
state = hass.states.get(f"{domain}.test")
assert state.state == STATE_UNAVAILABLE
if state_topic:
async_fire_mqtt_message(hass, state_topic, state_message)
state = hass.states.get(f"{domain}.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "online")
state = hass.states.get(f"{domain}.test")
assert state.state != STATE_UNAVAILABLE
async def help_test_custom_availability_payload(
hass,
mqtt_mock,
domain,
config,
no_assumed_state=False,
state_topic=None,
state_message=None,
):
"""Test availability by custom payload with defined topic.
This is a test helper for the MqttAvailability mixin.
"""
# Add availability settings to config
config = copy.deepcopy(config)
config[domain]["availability_topic"] = "availability-topic"
config[domain]["payload_available"] = "good"
config[domain]["payload_not_available"] = "nogood"
assert await async_setup_component(hass, domain, config,)
state = hass.states.get(f"{domain}.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "good")
state = hass.states.get(f"{domain}.test")
assert state.state != STATE_UNAVAILABLE
if no_assumed_state:
assert not state.attributes.get(ATTR_ASSUMED_STATE)
async_fire_mqtt_message(hass, "availability-topic", "nogood")
state = hass.states.get(f"{domain}.test")
assert state.state == STATE_UNAVAILABLE
if state_topic:
async_fire_mqtt_message(hass, state_topic, state_message)
state = hass.states.get(f"{domain}.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "good")
state = hass.states.get(f"{domain}.test")
assert state.state != STATE_UNAVAILABLE
async def help_test_setting_attribute_via_mqtt_json_message(
hass, mqtt_mock, domain, config
@ -21,6 +140,9 @@ async def help_test_setting_attribute_via_mqtt_json_message(
This is a test helper for the MqttAttributes mixin.
"""
# Add JSON attributes settings to config
config = copy.deepcopy(config)
config[domain]["json_attributes_topic"] = "attr-topic"
assert await async_setup_component(hass, domain, config,)
async_fire_mqtt_message(hass, "attr-topic", '{ "val": "100" }')
@ -29,6 +151,26 @@ async def help_test_setting_attribute_via_mqtt_json_message(
assert state.attributes.get("val") == "100"
async def help_test_setting_attribute_with_template(hass, mqtt_mock, domain, config):
"""Test the setting of attribute via MQTT with JSON payload.
This is a test helper for the MqttAttributes mixin.
"""
# Add JSON attributes settings to config
config = copy.deepcopy(config)
config[domain]["json_attributes_topic"] = "attr-topic"
config[domain]["json_attributes_template"] = "{{ value_json['Timer1'] | tojson }}"
assert await async_setup_component(hass, domain, config,)
async_fire_mqtt_message(
hass, "attr-topic", json.dumps({"Timer1": {"Arm": 0, "Time": "22:18"}})
)
state = hass.states.get(f"{domain}.test")
assert state.attributes.get("Arm") == 0
assert state.attributes.get("Time") == "22:18"
async def help_test_update_with_json_attrs_not_dict(
hass, mqtt_mock, caplog, domain, config
):
@ -36,6 +178,9 @@ async def help_test_update_with_json_attrs_not_dict(
This is a test helper for the MqttAttributes mixin.
"""
# Add JSON attributes settings to config
config = copy.deepcopy(config)
config[domain]["json_attributes_topic"] = "attr-topic"
assert await async_setup_component(hass, domain, config,)
async_fire_mqtt_message(hass, "attr-topic", '[ "list", "of", "things"]')
@ -52,6 +197,9 @@ async def help_test_update_with_json_attrs_bad_JSON(
This is a test helper for the MqttAttributes mixin.
"""
# Add JSON attributes settings to config
config = copy.deepcopy(config)
config[domain]["json_attributes_topic"] = "attr-topic"
assert await async_setup_component(hass, domain, config,)
async_fire_mqtt_message(hass, "attr-topic", "This is not JSON")
@ -61,13 +209,19 @@ async def help_test_update_with_json_attrs_bad_JSON(
assert "Erroneous JSON: This is not JSON" in caplog.text
async def help_test_discovery_update_attr(
hass, mqtt_mock, caplog, domain, data1, data2
):
async def help_test_discovery_update_attr(hass, mqtt_mock, caplog, domain, config):
"""Test update of discovered MQTTAttributes.
This is a test helper for the MqttAttributes mixin.
"""
# Add JSON attributes settings to config
config1 = copy.deepcopy(config)
config1[domain]["json_attributes_topic"] = "attr-topic1"
config2 = copy.deepcopy(config)
config2[domain]["json_attributes_topic"] = "attr-topic2"
data1 = json.dumps(config1[domain])
data2 = json.dumps(config2[domain])
entry = MockConfigEntry(domain=mqtt.DOMAIN)
await async_start(hass, "homeassistant", {}, entry)
async_fire_mqtt_message(hass, f"homeassistant/{domain}/bla/config", data1)
@ -173,6 +327,11 @@ async def help_test_entity_device_info_with_identifier(hass, mqtt_mock, domain,
This is a test helper for the MqttDiscoveryUpdate mixin.
"""
# Add device settings to config
config = copy.deepcopy(config[domain])
config["device"] = copy.deepcopy(DEFAULT_CONFIG_DEVICE_INFO_ID)
config["unique_id"] = "veryunique"
entry = MockConfigEntry(domain=mqtt.DOMAIN)
entry.add_to_hass(hass)
await async_start(hass, "homeassistant", {}, entry)
@ -185,6 +344,33 @@ async def help_test_entity_device_info_with_identifier(hass, mqtt_mock, domain,
device = registry.async_get_device({("mqtt", "helloworld")}, set())
assert device is not None
assert device.identifiers == {("mqtt", "helloworld")}
assert device.manufacturer == "Whatever"
assert device.name == "Beer"
assert device.model == "Glass"
assert device.sw_version == "0.1-beta"
async def help_test_entity_device_info_with_connection(hass, mqtt_mock, domain, config):
"""Test device registry integration.
This is a test helper for the MqttDiscoveryUpdate mixin.
"""
# Add device settings to config
config = copy.deepcopy(config[domain])
config["device"] = copy.deepcopy(DEFAULT_CONFIG_DEVICE_INFO_MAC)
config["unique_id"] = "veryunique"
entry = MockConfigEntry(domain=mqtt.DOMAIN)
entry.add_to_hass(hass)
await async_start(hass, "homeassistant", {}, entry)
registry = await hass.helpers.device_registry.async_get_registry()
data = json.dumps(config)
async_fire_mqtt_message(hass, f"homeassistant/{domain}/bla/config", data)
await hass.async_block_till_done()
device = registry.async_get_device(set(), {("mac", "02:5b:26:a8:dc:12")})
assert device is not None
assert device.connections == {("mac", "02:5b:26:a8:dc:12")}
assert device.manufacturer == "Whatever"
assert device.name == "Beer"
@ -194,6 +380,11 @@ async def help_test_entity_device_info_with_identifier(hass, mqtt_mock, domain,
async def help_test_entity_device_info_remove(hass, mqtt_mock, domain, config):
"""Test device registry remove."""
# Add device settings to config
config = copy.deepcopy(config[domain])
config["device"] = copy.deepcopy(DEFAULT_CONFIG_DEVICE_INFO_ID)
config["unique_id"] = "veryunique"
entry = MockConfigEntry(domain=mqtt.DOMAIN)
entry.add_to_hass(hass)
await async_start(hass, "homeassistant", {}, entry)
@ -221,6 +412,11 @@ async def help_test_entity_device_info_update(hass, mqtt_mock, domain, config):
This is a test helper for the MqttDiscoveryUpdate mixin.
"""
# Add device settings to config
config = copy.deepcopy(config[domain])
config["device"] = copy.deepcopy(DEFAULT_CONFIG_DEVICE_INFO_ID)
config["unique_id"] = "veryunique"
entry = MockConfigEntry(domain=mqtt.DOMAIN)
entry.add_to_hass(hass)
await async_start(hass, "homeassistant", {}, entry)
@ -244,27 +440,36 @@ async def help_test_entity_device_info_update(hass, mqtt_mock, domain, config):
assert device.name == "Milk"
async def help_test_entity_id_update(hass, mqtt_mock, domain, config):
async def help_test_entity_id_update(hass, mqtt_mock, domain, config, topics=None):
"""Test MQTT subscriptions are managed when entity_id is updated."""
# Add unique_id to config
config = copy.deepcopy(config)
config[domain]["unique_id"] = "TOTALLY_UNIQUE"
if topics is None:
# Add default topics to config
config[domain]["availability_topic"] = "avty-topic"
config[domain]["state_topic"] = "test-topic"
topics = ["avty-topic", "test-topic"]
assert len(topics) > 0
registry = mock_registry(hass, {})
mock_mqtt = await async_mock_mqtt_component(hass)
assert await async_setup_component(hass, domain, config,)
state = hass.states.get(f"{domain}.beer")
state = hass.states.get(f"{domain}.test")
assert state is not None
assert mock_mqtt.async_subscribe.call_count == 2
mock_mqtt.async_subscribe.assert_any_call("test-topic", ANY, 0, "utf-8")
mock_mqtt.async_subscribe.assert_any_call("avty-topic", ANY, 0, "utf-8")
assert mock_mqtt.async_subscribe.call_count == len(topics)
for topic in topics:
mock_mqtt.async_subscribe.assert_any_call(topic, ANY, ANY, ANY)
mock_mqtt.async_subscribe.reset_mock()
registry.async_update_entity(f"{domain}.beer", new_entity_id=f"{domain}.milk")
registry.async_update_entity(f"{domain}.test", new_entity_id=f"{domain}.milk")
await hass.async_block_till_done()
state = hass.states.get(f"{domain}.beer")
state = hass.states.get(f"{domain}.test")
assert state is None
state = hass.states.get(f"{domain}.milk")
assert state is not None
assert mock_mqtt.async_subscribe.call_count == 2
mock_mqtt.async_subscribe.assert_any_call("test-topic", ANY, 0, "utf-8")
mock_mqtt.async_subscribe.assert_any_call("avty-topic", ANY, 0, "utf-8")
for topic in topics:
mock_mqtt.async_subscribe.assert_any_call(topic, ANY, ANY, ANY)

152
tests/components/mqtt/test_alarm_control_panel.py
View File

@ -10,20 +10,24 @@ from homeassistant.const import (
STATE_ALARM_DISARMED,
STATE_ALARM_PENDING,
STATE_ALARM_TRIGGERED,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from .common import (
help_test_availability_without_topic,
help_test_custom_availability_payload,
help_test_default_availability_payload,
help_test_discovery_broken,
help_test_discovery_removal,
help_test_discovery_update,
help_test_discovery_update_attr,
help_test_entity_device_info_remove,
help_test_entity_device_info_update,
help_test_entity_device_info_with_connection,
help_test_entity_device_info_with_identifier,
help_test_entity_id_update,
help_test_setting_attribute_via_mqtt_json_message,
help_test_setting_attribute_with_template,
help_test_unique_id,
help_test_update_with_json_attrs_bad_JSON,
help_test_update_with_json_attrs_not_dict,
@ -47,16 +51,6 @@ DEFAULT_CONFIG = {
}
}
DEFAULT_CONFIG_ATTR = {
alarm_control_panel.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "alarm/state",
"command_topic": "alarm/command",
"json_attributes_topic": "attr-topic",
}
}
DEFAULT_CONFIG_CODE = {
alarm_control_panel.DOMAIN: {
"platform": "mqtt",
@ -68,22 +62,6 @@ DEFAULT_CONFIG_CODE = {
}
}
DEFAULT_CONFIG_DEVICE_INFO = {
"platform": "mqtt",
"name": "test",
"state_topic": "test-topic",
"command_topic": "test-command-topic",
"device": {
"identifiers": ["helloworld"],
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
},
"unique_id": "veryunique",
}
async def test_fail_setup_without_state_topic(hass, mqtt_mock):
"""Test for failing with no state topic."""
@ -331,48 +309,6 @@ async def test_disarm_not_publishes_mqtt_with_invalid_code_when_req(hass, mqtt_m
assert mqtt_mock.async_publish.call_count == call_count
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
config = copy.deepcopy(DEFAULT_CONFIG_CODE)
config[alarm_control_panel.DOMAIN]["availability_topic"] = "availability-topic"
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, config,)
state = hass.states.get("alarm_control_panel.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "online")
state = hass.states.get("alarm_control_panel.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "offline")
state = hass.states.get("alarm_control_panel.test")
assert state.state == STATE_UNAVAILABLE
async def test_custom_availability_payload(hass, mqtt_mock):
"""Test availability by custom payload with defined topic."""
config = copy.deepcopy(DEFAULT_CONFIG)
config[alarm_control_panel.DOMAIN]["availability_topic"] = "availability-topic"
config[alarm_control_panel.DOMAIN]["payload_available"] = "good"
config[alarm_control_panel.DOMAIN]["payload_not_available"] = "nogood"
assert await async_setup_component(hass, alarm_control_panel.DOMAIN, config,)
state = hass.states.get("alarm_control_panel.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "good")
state = hass.states.get("alarm_control_panel.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "nogood")
state = hass.states.get("alarm_control_panel.test")
assert state.state == STATE_UNAVAILABLE
async def test_update_state_via_state_topic_template(hass, mqtt_mock):
"""Test updating with template_value via state topic."""
assert await async_setup_component(
@ -403,38 +339,59 @@ async def test_update_state_via_state_topic_template(hass, mqtt_mock):
assert state.state == STATE_ALARM_ARMED_AWAY
async def test_availability_without_topic(hass, mqtt_mock):
"""Test availability without defined availability topic."""
await help_test_availability_without_topic(
hass, mqtt_mock, alarm_control_panel.DOMAIN, DEFAULT_CONFIG_CODE
)
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
await help_test_default_availability_payload(
hass, mqtt_mock, alarm_control_panel.DOMAIN, DEFAULT_CONFIG_CODE
)
async def test_custom_availability_payload(hass, mqtt_mock):
"""Test availability by custom payload with defined topic."""
await help_test_custom_availability_payload(
hass, mqtt_mock, alarm_control_panel.DOMAIN, DEFAULT_CONFIG_CODE
)
async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_via_mqtt_json_message(
hass, mqtt_mock, alarm_control_panel.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, alarm_control_panel.DOMAIN, DEFAULT_CONFIG
)
async def test_setting_attribute_with_template(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_with_template(
hass, mqtt_mock, alarm_control_panel.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_not_dict(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_not_dict(
hass, mqtt_mock, caplog, alarm_control_panel.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, alarm_control_panel.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_bad_JSON(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_bad_JSON(
hass, mqtt_mock, caplog, alarm_control_panel.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, alarm_control_panel.DOMAIN, DEFAULT_CONFIG
)
async def test_discovery_update_attr(hass, mqtt_mock, caplog):
"""Test update of discovered MQTTAttributes."""
config1 = copy.deepcopy(DEFAULT_CONFIG[alarm_control_panel.DOMAIN])
config2 = copy.deepcopy(DEFAULT_CONFIG[alarm_control_panel.DOMAIN])
config1["json_attributes_topic"] = "attr-topic1"
config2["json_attributes_topic"] = "attr-topic2"
data1 = json.dumps(config1)
data2 = json.dumps(config2)
await help_test_discovery_update_attr(
hass, mqtt_mock, caplog, alarm_control_panel.DOMAIN, data1, data2
hass, mqtt_mock, caplog, alarm_control_panel.DOMAIN, DEFAULT_CONFIG
)
@ -496,49 +453,36 @@ async def test_discovery_broken(hass, mqtt_mock, caplog):
)
async def test_entity_device_info_with_connection(hass, mqtt_mock):
"""Test MQTT alarm control panel device registry integration."""
await help_test_entity_device_info_with_connection(
hass, mqtt_mock, alarm_control_panel.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_with_identifier(hass, mqtt_mock):
"""Test MQTT alarm control panel device registry integration."""
await help_test_entity_device_info_with_identifier(
hass, mqtt_mock, alarm_control_panel.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, alarm_control_panel.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_update(hass, mqtt_mock):
"""Test device registry update."""
await help_test_entity_device_info_update(
hass, mqtt_mock, alarm_control_panel.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, alarm_control_panel.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_remove(hass, mqtt_mock):
"""Test device registry remove."""
config = {
"platform": "mqtt",
"name": "Test 1",
"state_topic": "test-topic",
"command_topic": "test-command-topic",
"device": {"identifiers": ["helloworld"]},
"unique_id": "veryunique",
}
await help_test_entity_device_info_remove(
hass, mqtt_mock, alarm_control_panel.DOMAIN, config
hass, mqtt_mock, alarm_control_panel.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_id_update(hass, mqtt_mock):
"""Test MQTT subscriptions are managed when entity_id is updated."""
config = {
alarm_control_panel.DOMAIN: [
{
"platform": "mqtt",
"name": "beer",
"state_topic": "test-topic",
"command_topic": "command-topic",
"availability_topic": "avty-topic",
"unique_id": "TOTALLY_UNIQUE",
}
]
}
await help_test_entity_id_update(
hass, mqtt_mock, alarm_control_panel.DOMAIN, config
hass, mqtt_mock, alarm_control_panel.DOMAIN, DEFAULT_CONFIG
)

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

@ -16,15 +16,20 @@ from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from .common import (
help_test_availability_without_topic,
help_test_custom_availability_payload,
help_test_default_availability_payload,
help_test_discovery_broken,
help_test_discovery_removal,
help_test_discovery_update,
help_test_discovery_update_attr,
help_test_entity_device_info_remove,
help_test_entity_device_info_update,
help_test_entity_device_info_with_connection,
help_test_entity_device_info_with_identifier,
help_test_entity_id_update,
help_test_setting_attribute_via_mqtt_json_message,
help_test_setting_attribute_with_template,
help_test_unique_id,
help_test_update_with_json_attrs_bad_JSON,
help_test_update_with_json_attrs_not_dict,
@ -40,30 +45,6 @@ DEFAULT_CONFIG = {
}
}
DEFAULT_CONFIG_ATTR = {
binary_sensor.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "test-topic",
"json_attributes_topic": "attr-topic",
}
}
DEFAULT_CONFIG_DEVICE_INFO = {
"platform": "mqtt",
"name": "test",
"state_topic": "test-topic",
"device": {
"identifiers": ["helloworld"],
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
},
"unique_id": "veryunique",
}
async def test_setting_sensor_value_expires_availability_topic(hass, mqtt_mock, caplog):
"""Test the expiration of the value."""
@ -265,81 +246,24 @@ async def test_invalid_device_class(hass, mqtt_mock):
async def test_availability_without_topic(hass, mqtt_mock):
"""Test availability without defined availability topic."""
assert await async_setup_component(
hass,
binary_sensor.DOMAIN,
{
binary_sensor.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "state-topic",
}
},
await help_test_availability_without_topic(
hass, mqtt_mock, binary_sensor.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("binary_sensor.test")
assert state.state != STATE_UNAVAILABLE
async def test_availability_by_defaults(hass, mqtt_mock):
"""Test availability by defaults with defined topic."""
assert await async_setup_component(
hass,
binary_sensor.DOMAIN,
{
binary_sensor.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "state-topic",
"availability_topic": "availability-topic",
}
},
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
await help_test_default_availability_payload(
hass, mqtt_mock, binary_sensor.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("binary_sensor.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "online")
state = hass.states.get("binary_sensor.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "offline")
state = hass.states.get("binary_sensor.test")
assert state.state == STATE_UNAVAILABLE
async def test_availability_by_custom_payload(hass, mqtt_mock):
async def test_custom_availability_payload(hass, mqtt_mock):
"""Test availability by custom payload with defined topic."""
assert await async_setup_component(
hass,
binary_sensor.DOMAIN,
{
binary_sensor.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "state-topic",
"availability_topic": "availability-topic",
"payload_available": "good",
"payload_not_available": "nogood",
}
},
await help_test_custom_availability_payload(
hass, mqtt_mock, binary_sensor.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("binary_sensor.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "good")
state = hass.states.get("binary_sensor.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "nogood")
state = hass.states.get("binary_sensor.test")
assert state.state == STATE_UNAVAILABLE
async def test_force_update_disabled(hass, mqtt_mock):
"""Test force update option."""
@ -459,35 +383,35 @@ async def test_off_delay(hass, mqtt_mock):
async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_via_mqtt_json_message(
hass, mqtt_mock, binary_sensor.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, binary_sensor.DOMAIN, DEFAULT_CONFIG
)
async def test_setting_attribute_with_template(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_with_template(
hass, mqtt_mock, binary_sensor.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_not_dict(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_not_dict(
hass, mqtt_mock, caplog, binary_sensor.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, binary_sensor.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_bad_JSON(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_bad_JSON(
hass, mqtt_mock, caplog, binary_sensor.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, binary_sensor.DOMAIN, DEFAULT_CONFIG
)
async def test_discovery_update_attr(hass, mqtt_mock, caplog):
"""Test update of discovered MQTTAttributes."""
config1 = copy.deepcopy(DEFAULT_CONFIG_ATTR[binary_sensor.DOMAIN])
config2 = copy.deepcopy(DEFAULT_CONFIG_ATTR[binary_sensor.DOMAIN])
config1["json_attributes_topic"] = "attr-topic1"
config2["json_attributes_topic"] = "attr-topic2"
data1 = json.dumps(config1)
data2 = json.dumps(config2)
await help_test_discovery_update_attr(
hass, mqtt_mock, caplog, binary_sensor.DOMAIN, data1, data2
hass, mqtt_mock, caplog, binary_sensor.DOMAIN, DEFAULT_CONFIG
)
@ -543,45 +467,36 @@ async def test_discovery_broken(hass, mqtt_mock, caplog):
)
async def test_entity_device_info_with_connection(hass, mqtt_mock):
"""Test MQTT binary sensor device registry integration."""
await help_test_entity_device_info_with_connection(
hass, mqtt_mock, binary_sensor.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_with_identifier(hass, mqtt_mock):
"""Test MQTT binary sensor device registry integration."""
await help_test_entity_device_info_with_identifier(
hass, mqtt_mock, binary_sensor.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, binary_sensor.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_update(hass, mqtt_mock):
"""Test device registry update."""
await help_test_entity_device_info_update(
hass, mqtt_mock, binary_sensor.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, binary_sensor.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_remove(hass, mqtt_mock):
"""Test device registry remove."""
config = {
"platform": "mqtt",
"name": "Test 1",
"state_topic": "test-topic",
"device": {"identifiers": ["helloworld"]},
"unique_id": "veryunique",
}
await help_test_entity_device_info_remove(
hass, mqtt_mock, binary_sensor.DOMAIN, config
hass, mqtt_mock, binary_sensor.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_id_update(hass, mqtt_mock):
"""Test MQTT subscriptions are managed when entity_id is updated."""
config = {
binary_sensor.DOMAIN: [
{
"platform": "mqtt",
"name": "beer",
"state_topic": "test-topic",
"availability_topic": "avty-topic",
"unique_id": "TOTALLY_UNIQUE",
}
]
}
await help_test_entity_id_update(hass, mqtt_mock, binary_sensor.DOMAIN, config)
await help_test_entity_id_update(
hass, mqtt_mock, binary_sensor.DOMAIN, DEFAULT_CONFIG
)

229
tests/components/mqtt/test_camera.py
View File

@ -1,18 +1,32 @@
"""The tests for mqtt camera component."""
import json
from unittest.mock import ANY
from homeassistant.components import camera, mqtt
from homeassistant.components.mqtt.discovery import async_start
from homeassistant.setup import async_setup_component
from .common import (
help_test_discovery_broken,
help_test_discovery_removal,
help_test_discovery_update,
help_test_entity_device_info_remove,
help_test_entity_device_info_update,
help_test_entity_device_info_with_connection,
help_test_entity_device_info_with_identifier,
help_test_entity_id_update,
help_test_unique_id,
)
from tests.common import (
MockConfigEntry,
async_fire_mqtt_message,
async_mock_mqtt_component,
mock_registry,
)
DEFAULT_CONFIG = {
camera.DOMAIN: {"platform": "mqtt", "name": "test", "topic": "test_topic"}
}
async def test_run_camera_setup(hass, aiohttp_client):
"""Test that it fetches the given payload."""
@ -37,51 +51,29 @@ async def test_run_camera_setup(hass, aiohttp_client):
async def test_unique_id(hass):
"""Test unique id option only creates one camera per unique_id."""
await async_mock_mqtt_component(hass)
await async_setup_component(
hass,
"camera",
{
"camera": [
{
"platform": "mqtt",
"name": "Test Camera 1",
"topic": "test-topic",
"unique_id": "TOTALLY_UNIQUE",
},
{
"platform": "mqtt",
"name": "Test Camera 2",
"topic": "test-topic",
"unique_id": "TOTALLY_UNIQUE",
},
]
},
)
async_fire_mqtt_message(hass, "test-topic", "payload")
assert len(hass.states.async_all()) == 1
config = {
camera.DOMAIN: [
{
"platform": "mqtt",
"name": "Test 1",
"topic": "test-topic",
"unique_id": "TOTALLY_UNIQUE",
},
{
"platform": "mqtt",
"name": "Test 2",
"topic": "test-topic",
"unique_id": "TOTALLY_UNIQUE",
},
]
}
await help_test_unique_id(hass, camera.DOMAIN, config)
async def test_discovery_removal_camera(hass, mqtt_mock, caplog):
"""Test removal of discovered camera."""
entry = MockConfigEntry(domain=mqtt.DOMAIN)
await async_start(hass, "homeassistant", {}, entry)
data = '{ "name": "Beer",' ' "topic": "test_topic"}'
async_fire_mqtt_message(hass, "homeassistant/camera/bla/config", data)
await hass.async_block_till_done()
state = hass.states.get("camera.beer")
assert state is not None
assert state.name == "Beer"
async_fire_mqtt_message(hass, "homeassistant/camera/bla/config", "")
await hass.async_block_till_done()
state = hass.states.get("camera.beer")
assert state is None
data = json.dumps(DEFAULT_CONFIG[camera.DOMAIN])
await help_test_discovery_removal(hass, mqtt_mock, caplog, camera.DOMAIN, data)
async def test_discovery_update_camera(hass, mqtt_mock, caplog):
@ -92,21 +84,9 @@ async def test_discovery_update_camera(hass, mqtt_mock, caplog):
data1 = '{ "name": "Beer",' ' "topic": "test_topic"}'
data2 = '{ "name": "Milk",' ' "topic": "test_topic"}'
async_fire_mqtt_message(hass, "homeassistant/camera/bla/config", data1)
await hass.async_block_till_done()
state = hass.states.get("camera.beer")
assert state is not None
assert state.name == "Beer"
async_fire_mqtt_message(hass, "homeassistant/camera/bla/config", data2)
await hass.async_block_till_done()
state = hass.states.get("camera.beer")
assert state is not None
assert state.name == "Milk"
state = hass.states.get("camera.milk")
assert state is None
await help_test_discovery_update(
hass, mqtt_mock, caplog, camera.DOMAIN, data1, data2
)
async def test_discovery_broken(hass, mqtt_mock, caplog):
@ -117,130 +97,41 @@ async def test_discovery_broken(hass, mqtt_mock, caplog):
data1 = '{ "name": "Beer" }'
data2 = '{ "name": "Milk",' ' "topic": "test_topic"}'
async_fire_mqtt_message(hass, "homeassistant/camera/bla/config", data1)
await hass.async_block_till_done()
state = hass.states.get("camera.beer")
assert state is None
async_fire_mqtt_message(hass, "homeassistant/camera/bla/config", data2)
await hass.async_block_till_done()
state = hass.states.get("camera.milk")
assert state is not None
assert state.name == "Milk"
state = hass.states.get("camera.beer")
assert state is None
async def test_entity_id_update(hass, mqtt_mock):
"""Test MQTT subscriptions are managed when entity_id is updated."""
registry = mock_registry(hass, {})
mock_mqtt = await async_mock_mqtt_component(hass)
assert await async_setup_component(
hass,
camera.DOMAIN,
{
camera.DOMAIN: [
{
"platform": "mqtt",
"name": "beer",
"topic": "test-topic",
"unique_id": "TOTALLY_UNIQUE",
}
]
},
await help_test_discovery_broken(
hass, mqtt_mock, caplog, camera.DOMAIN, data1, data2
)
state = hass.states.get("camera.beer")
assert state is not None
assert mock_mqtt.async_subscribe.call_count == 1
mock_mqtt.async_subscribe.assert_any_call("test-topic", ANY, 0, None)
mock_mqtt.async_subscribe.reset_mock()
registry.async_update_entity("camera.beer", new_entity_id="camera.milk")
await hass.async_block_till_done()
state = hass.states.get("camera.beer")
assert state is None
state = hass.states.get("camera.milk")
assert state is not None
assert mock_mqtt.async_subscribe.call_count == 1
mock_mqtt.async_subscribe.assert_any_call("test-topic", ANY, 0, None)
async def test_entity_device_info_with_connection(hass, mqtt_mock):
"""Test MQTT camera device registry integration."""
await help_test_entity_device_info_with_connection(
hass, mqtt_mock, camera.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_with_identifier(hass, mqtt_mock):
"""Test MQTT camera device registry integration."""
entry = MockConfigEntry(domain=mqtt.DOMAIN)
entry.add_to_hass(hass)
await async_start(hass, "homeassistant", {}, entry)
registry = await hass.helpers.device_registry.async_get_registry()
data = json.dumps(
{
"platform": "mqtt",
"name": "Test 1",
"topic": "test-topic",
"device": {
"identifiers": ["helloworld"],
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
},
"unique_id": "veryunique",
}
await help_test_entity_device_info_with_identifier(
hass, mqtt_mock, camera.DOMAIN, DEFAULT_CONFIG
)
async_fire_mqtt_message(hass, "homeassistant/camera/bla/config", data)
await hass.async_block_till_done()
device = registry.async_get_device({("mqtt", "helloworld")}, set())
assert device is not None
assert device.identifiers == {("mqtt", "helloworld")}
assert device.connections == {("mac", "02:5b:26:a8:dc:12")}
assert device.manufacturer == "Whatever"
assert device.name == "Beer"
assert device.model == "Glass"
assert device.sw_version == "0.1-beta"
async def test_entity_device_info_update(hass, mqtt_mock):
"""Test device registry update."""
entry = MockConfigEntry(domain=mqtt.DOMAIN)
entry.add_to_hass(hass)
await async_start(hass, "homeassistant", {}, entry)
registry = await hass.helpers.device_registry.async_get_registry()
await help_test_entity_device_info_update(
hass, mqtt_mock, camera.DOMAIN, DEFAULT_CONFIG
)
config = {
"platform": "mqtt",
"name": "Test 1",
"topic": "test-topic",
"device": {
"identifiers": ["helloworld"],
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
},
"unique_id": "veryunique",
}
data = json.dumps(config)
async_fire_mqtt_message(hass, "homeassistant/camera/bla/config", data)
await hass.async_block_till_done()
async def test_entity_device_info_remove(hass, mqtt_mock):
"""Test device registry remove."""
await help_test_entity_device_info_remove(
hass, mqtt_mock, camera.DOMAIN, DEFAULT_CONFIG
)
device = registry.async_get_device({("mqtt", "helloworld")}, set())
assert device is not None
assert device.name == "Beer"
config["device"]["name"] = "Milk"
data = json.dumps(config)
async_fire_mqtt_message(hass, "homeassistant/camera/bla/config", data)
await hass.async_block_till_done()
device = registry.async_get_device({("mqtt", "helloworld")}, set())
assert device is not None
assert device.name == "Milk"
async def test_entity_id_update(hass, mqtt_mock):
"""Test MQTT subscriptions are managed when entity_id is updated."""
await help_test_entity_id_update(
hass, mqtt_mock, camera.DOMAIN, DEFAULT_CONFIG, ["test_topic"]
)

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

@ -23,18 +23,23 @@ from homeassistant.components.climate.const import (
SUPPORT_TARGET_TEMPERATURE,
SUPPORT_TARGET_TEMPERATURE_RANGE,
)
from homeassistant.const import STATE_OFF, STATE_UNAVAILABLE
from homeassistant.const import STATE_OFF
from .common import (
help_test_availability_without_topic,
help_test_custom_availability_payload,
help_test_default_availability_payload,
help_test_discovery_broken,
help_test_discovery_removal,
help_test_discovery_update,
help_test_discovery_update_attr,
help_test_entity_device_info_remove,
help_test_entity_device_info_update,
help_test_entity_device_info_with_connection,
help_test_entity_device_info_with_identifier,
help_test_entity_id_update,
help_test_setting_attribute_via_mqtt_json_message,
help_test_setting_attribute_with_template,
help_test_unique_id,
help_test_update_with_json_attrs_bad_JSON,
help_test_update_with_json_attrs_not_dict,
@ -46,7 +51,7 @@ from tests.components.climate import common
ENTITY_CLIMATE = "climate.test"
DEFAULT_CONFIG = {
"climate": {
CLIMATE_DOMAIN: {
"platform": "mqtt",
"name": "test",
"mode_command_topic": "mode-topic",
@ -61,32 +66,6 @@ DEFAULT_CONFIG = {
}
}
DEFAULT_CONFIG_ATTR = {
CLIMATE_DOMAIN: {
"platform": "mqtt",
"name": "test",
"power_state_topic": "test-topic",
"power_command_topic": "test_topic",
"json_attributes_topic": "attr-topic",
}
}
DEFAULT_CONFIG_DEVICE_INFO = {
"platform": "mqtt",
"name": "Test 1",
"power_state_topic": "test-topic",
"power_command_topic": "test-command-topic",
"device": {
"identifiers": ["helloworld"],
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
},
"unique_id": "veryunique",
}
async def test_setup_params(hass, mqtt_mock):
"""Test the initial parameters."""
@ -597,27 +576,25 @@ async def test_set_aux(hass, mqtt_mock):
assert state.attributes.get("aux_heat") == "off"
async def test_availability_without_topic(hass, mqtt_mock):
"""Test availability without defined availability topic."""
await help_test_availability_without_topic(
hass, mqtt_mock, CLIMATE_DOMAIN, DEFAULT_CONFIG
)
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
await help_test_default_availability_payload(
hass, mqtt_mock, CLIMATE_DOMAIN, DEFAULT_CONFIG
)
async def test_custom_availability_payload(hass, mqtt_mock):
"""Test availability by custom payload with defined topic."""
config = copy.deepcopy(DEFAULT_CONFIG)
config["climate"]["availability_topic"] = "availability-topic"
config["climate"]["payload_available"] = "good"
config["climate"]["payload_not_available"] = "nogood"
assert await async_setup_component(hass, CLIMATE_DOMAIN, config)
state = hass.states.get("climate.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "good")
state = hass.states.get("climate.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "nogood")
state = hass.states.get("climate.test")
assert state.state == STATE_UNAVAILABLE
await help_test_custom_availability_payload(
hass, mqtt_mock, CLIMATE_DOMAIN, DEFAULT_CONFIG
)
async def test_set_target_temperature_low_high_with_templates(hass, mqtt_mock, caplog):
@ -801,35 +778,35 @@ async def test_temp_step_custom(hass, mqtt_mock):
async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_via_mqtt_json_message(
hass, mqtt_mock, CLIMATE_DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, CLIMATE_DOMAIN, DEFAULT_CONFIG
)
async def test_setting_attribute_with_template(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_with_template(
hass, mqtt_mock, CLIMATE_DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_not_dict(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_not_dict(
hass, mqtt_mock, caplog, CLIMATE_DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, CLIMATE_DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_bad_JSON(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_bad_JSON(
hass, mqtt_mock, caplog, CLIMATE_DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, CLIMATE_DOMAIN, DEFAULT_CONFIG
)
async def test_discovery_update_attr(hass, mqtt_mock, caplog):
"""Test update of discovered MQTTAttributes."""
config1 = copy.deepcopy(DEFAULT_CONFIG[CLIMATE_DOMAIN])
config2 = copy.deepcopy(DEFAULT_CONFIG[CLIMATE_DOMAIN])
config1["json_attributes_topic"] = "attr-topic1"
config2["json_attributes_topic"] = "attr-topic2"
data1 = json.dumps(config1)
data2 = json.dumps(config2)
await help_test_discovery_update_attr(
hass, mqtt_mock, caplog, CLIMATE_DOMAIN, data1, data2
hass, mqtt_mock, caplog, CLIMATE_DOMAIN, DEFAULT_CONFIG
)
@ -880,47 +857,47 @@ async def test_discovery_broken(hass, mqtt_mock, caplog):
)
async def test_entity_device_info_with_connection(hass, mqtt_mock):
"""Test MQTT climate device registry integration."""
await help_test_entity_device_info_with_connection(
hass, mqtt_mock, CLIMATE_DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_with_identifier(hass, mqtt_mock):
"""Test MQTT climate device registry integration."""
await help_test_entity_device_info_with_identifier(
hass, mqtt_mock, CLIMATE_DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, CLIMATE_DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_update(hass, mqtt_mock):
"""Test device registry update."""
await help_test_entity_device_info_update(
hass, mqtt_mock, CLIMATE_DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, CLIMATE_DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_remove(hass, mqtt_mock):
"""Test device registry remove."""
config = {
"platform": "mqtt",
"name": "Test 1",
"power_state_topic": "test-topic",
"power_command_topic": "test-command-topic",
"device": {"identifiers": ["helloworld"]},
"unique_id": "veryunique",
}
await help_test_entity_device_info_remove(hass, mqtt_mock, CLIMATE_DOMAIN, config)
await help_test_entity_device_info_remove(
hass, mqtt_mock, CLIMATE_DOMAIN, DEFAULT_CONFIG
)
async def test_entity_id_update(hass, mqtt_mock):
"""Test MQTT subscriptions are managed when entity_id is updated."""
config = {
CLIMATE_DOMAIN: [
{
"platform": "mqtt",
"name": "beer",
"mode_state_topic": "test-topic",
"availability_topic": "avty-topic",
"unique_id": "TOTALLY_UNIQUE",
}
]
CLIMATE_DOMAIN: {
"platform": "mqtt",
"name": "test",
"mode_state_topic": "test-topic",
"availability_topic": "avty-topic",
}
}
await help_test_entity_id_update(hass, mqtt_mock, CLIMATE_DOMAIN, config)
await help_test_entity_id_update(
hass, mqtt_mock, CLIMATE_DOMAIN, config, ["test-topic", "avty-topic"]
)
async def test_precision_default(hass, mqtt_mock):

175
tests/components/mqtt/test_cover.py
View File

@ -18,21 +18,25 @@ from homeassistant.const import (
STATE_CLOSING,
STATE_OPEN,
STATE_OPENING,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.setup import async_setup_component
from .common import (
help_test_availability_without_topic,
help_test_custom_availability_payload,
help_test_default_availability_payload,
help_test_discovery_broken,
help_test_discovery_removal,
help_test_discovery_update,
help_test_discovery_update_attr,
help_test_entity_device_info_remove,
help_test_entity_device_info_update,
help_test_entity_device_info_with_connection,
help_test_entity_device_info_with_identifier,
help_test_entity_id_update,
help_test_setting_attribute_via_mqtt_json_message,
help_test_setting_attribute_with_template,
help_test_unique_id,
help_test_update_with_json_attrs_bad_JSON,
help_test_update_with_json_attrs_not_dict,
@ -40,29 +44,8 @@ from .common import (
from tests.common import async_fire_mqtt_message
DEFAULT_CONFIG_ATTR = {
cover.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "test-topic",
"json_attributes_topic": "attr-topic",
}
}
DEFAULT_CONFIG_DEVICE_INFO = {
"platform": "mqtt",
"name": "Test 1",
"state_topic": "test-topic",
"command_topic": "test-command-topic",
"device": {
"identifiers": ["helloworld"],
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
},
"unique_id": "veryunique",
DEFAULT_CONFIG = {
cover.DOMAIN: {"platform": "mqtt", "name": "test", "state_topic": "test-topic"}
}
@ -1601,88 +1584,24 @@ async def test_find_in_range_altered_inverted(hass, mqtt_mock):
async def test_availability_without_topic(hass, mqtt_mock):
"""Test availability without defined availability topic."""
assert await async_setup_component(
hass,
cover.DOMAIN,
{
cover.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "state-topic",
"command_topic": "command-topic",
}
},
await help_test_availability_without_topic(
hass, mqtt_mock, cover.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("cover.test")
assert state.state != STATE_UNAVAILABLE
async def test_availability_by_defaults(hass, mqtt_mock):
"""Test availability by defaults with defined topic."""
assert await async_setup_component(
hass,
cover.DOMAIN,
{
cover.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "state-topic",
"command_topic": "command-topic",
"availability_topic": "availability-topic",
}
},
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
await help_test_default_availability_payload(
hass, mqtt_mock, cover.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("cover.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "online")
await hass.async_block_till_done()
state = hass.states.get("cover.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "offline")
await hass.async_block_till_done()
state = hass.states.get("cover.test")
assert state.state == STATE_UNAVAILABLE
async def test_availability_by_custom_payload(hass, mqtt_mock):
async def test_custom_availability_payload(hass, mqtt_mock):
"""Test availability by custom payload with defined topic."""
assert await async_setup_component(
hass,
cover.DOMAIN,
{
cover.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "state-topic",
"command_topic": "command-topic",
"availability_topic": "availability-topic",
"payload_available": "good",
"payload_not_available": "nogood",
}
},
await help_test_custom_availability_payload(
hass, mqtt_mock, cover.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("cover.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "good")
await hass.async_block_till_done()
state = hass.states.get("cover.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "nogood")
await hass.async_block_till_done()
state = hass.states.get("cover.test")
assert state.state == STATE_UNAVAILABLE
async def test_valid_device_class(hass, mqtt_mock):
"""Test the setting of a valid sensor class."""
@ -1725,39 +1644,35 @@ async def test_invalid_device_class(hass, mqtt_mock):
async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_via_mqtt_json_message(
hass, mqtt_mock, cover.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, cover.DOMAIN, DEFAULT_CONFIG
)
async def test_setting_attribute_with_template(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_with_template(
hass, mqtt_mock, cover.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_not_dict(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_not_dict(
hass, mqtt_mock, caplog, cover.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, cover.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_bad_JSON(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_bad_JSON(
hass, mqtt_mock, caplog, cover.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, cover.DOMAIN, DEFAULT_CONFIG
)
async def test_discovery_update_attr(hass, mqtt_mock, caplog):
"""Test update of discovered MQTTAttributes."""
data1 = (
'{ "name": "test",'
' "command_topic": "test_topic",'
' "json_attributes_topic": "attr-topic1" }'
)
data2 = (
'{ "name": "test",'
' "command_topic": "test_topic",'
' "json_attributes_topic": "attr-topic2" }'
)
await help_test_discovery_update_attr(
hass, mqtt_mock, caplog, cover.DOMAIN, data1, data2
hass, mqtt_mock, caplog, cover.DOMAIN, DEFAULT_CONFIG
)
@ -1806,44 +1721,34 @@ async def test_discovery_broken(hass, mqtt_mock, caplog):
)
async def test_entity_device_info_with_connection(hass, mqtt_mock):
"""Test MQTT cover device registry integration."""
await help_test_entity_device_info_with_connection(
hass, mqtt_mock, cover.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_with_identifier(hass, mqtt_mock):
"""Test MQTT cover device registry integration."""
await help_test_entity_device_info_with_identifier(
hass, mqtt_mock, cover.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, cover.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_update(hass, mqtt_mock):
"""Test device registry update."""
await help_test_entity_device_info_update(
hass, mqtt_mock, cover.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, cover.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_remove(hass, mqtt_mock):
"""Test device registry remove."""
config = {
"platform": "mqtt",
"name": "Test 1",
"state_topic": "test-topic",
"command_topic": "test-command-topic",
"device": {"identifiers": ["helloworld"]},
"unique_id": "veryunique",
}
await help_test_entity_device_info_remove(hass, mqtt_mock, cover.DOMAIN, config)
await help_test_entity_device_info_remove(
hass, mqtt_mock, cover.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_id_update(hass, mqtt_mock):
"""Test MQTT subscriptions are managed when entity_id is updated."""
config = {
cover.DOMAIN: [
{
"platform": "mqtt",
"name": "beer",
"state_topic": "test-topic",
"availability_topic": "avty-topic",
"unique_id": "TOTALLY_UNIQUE",
}
]
}
await help_test_entity_id_update(hass, mqtt_mock, cover.DOMAIN, config)
await help_test_entity_id_update(hass, mqtt_mock, cover.DOMAIN, DEFAULT_CONFIG)

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

@ -757,6 +757,40 @@ async def test_attach_remove_late2(hass, device_reg, mqtt_mock):
assert len(calls) == 0
async def test_entity_device_info_with_connection(hass, mqtt_mock):
"""Test MQTT device registry integration."""
entry = MockConfigEntry(domain=DOMAIN)
entry.add_to_hass(hass)
await async_start(hass, "homeassistant", {}, entry)
registry = await hass.helpers.device_registry.async_get_registry()
data = json.dumps(
{
"automation_type": "trigger",
"topic": "test-topic",
"type": "foo",
"subtype": "bar",
"device": {
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
},
}
)
async_fire_mqtt_message(hass, "homeassistant/device_automation/bla/config", data)
await hass.async_block_till_done()
device = registry.async_get_device(set(), {("mac", "02:5b:26:a8:dc:12")})
assert device is not None
assert device.connections == {("mac", "02:5b:26:a8:dc:12")}
assert device.manufacturer == "Whatever"
assert device.name == "Beer"
assert device.model == "Glass"
assert device.sw_version == "0.1-beta"
async def test_entity_device_info_with_identifier(hass, mqtt_mock):
"""Test MQTT device registry integration."""
entry = MockConfigEntry(domain=DOMAIN)
@ -772,7 +806,6 @@ async def test_entity_device_info_with_identifier(hass, mqtt_mock):
"subtype": "bar",
"device": {
"identifiers": ["helloworld"],
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
@ -786,7 +819,6 @@ async def test_entity_device_info_with_identifier(hass, mqtt_mock):
device = registry.async_get_device({("mqtt", "helloworld")}, set())
assert device is not None
assert device.identifiers == {("mqtt", "helloworld")}
assert device.connections == {("mac", "02:5b:26:a8:dc:12")}
assert device.manufacturer == "Whatever"
assert device.name == "Beer"
assert device.model == "Glass"

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

@ -1,23 +1,23 @@
"""Test MQTT fans."""
from homeassistant.components import fan
from homeassistant.const import (
ATTR_ASSUMED_STATE,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.const import ATTR_ASSUMED_STATE, STATE_OFF, STATE_ON
from homeassistant.setup import async_setup_component
from .common import (
help_test_availability_without_topic,
help_test_custom_availability_payload,
help_test_default_availability_payload,
help_test_discovery_broken,
help_test_discovery_removal,
help_test_discovery_update,
help_test_discovery_update_attr,
help_test_entity_device_info_remove,
help_test_entity_device_info_update,
help_test_entity_device_info_with_connection,
help_test_entity_device_info_with_identifier,
help_test_entity_id_update,
help_test_setting_attribute_via_mqtt_json_message,
help_test_setting_attribute_with_template,
help_test_unique_id,
help_test_update_with_json_attrs_bad_JSON,
help_test_update_with_json_attrs_not_dict,
@ -26,31 +26,15 @@ from .common import (
from tests.common import async_fire_mqtt_message
from tests.components.fan import common
DEFAULT_CONFIG_ATTR = {
DEFAULT_CONFIG = {
fan.DOMAIN: {
"platform": "mqtt",
"name": "test",
"command_topic": "test-topic",
"json_attributes_topic": "attr-topic",
"state_topic": "state-topic",
"command_topic": "command-topic",
}
}
DEFAULT_CONFIG_DEVICE_INFO = {
"platform": "mqtt",
"name": "Test 1",
"state_topic": "test-topic",
"command_topic": "test-command-topic",
"device": {
"identifiers": ["helloworld"],
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
},
"unique_id": "veryunique",
}
async def test_fail_setup_if_no_command_topic(hass, mqtt_mock):
"""Test if command fails with command topic."""
@ -385,125 +369,59 @@ async def test_sending_mqtt_commands_and_explicit_optimistic(hass, mqtt_mock):
assert state.attributes.get(ATTR_ASSUMED_STATE)
async def test_default_availability_payload(hass, mqtt_mock):
"""Test the availability payload."""
assert await async_setup_component(
hass,
fan.DOMAIN,
{
fan.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "state-topic",
"command_topic": "command-topic",
"availability_topic": "availability_topic",
}
},
async def test_availability_without_topic(hass, mqtt_mock):
"""Test availability without defined availability topic."""
await help_test_availability_without_topic(
hass, mqtt_mock, fan.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("fan.test")
assert state.state is STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability_topic", "online")
state = hass.states.get("fan.test")
assert state.state is not STATE_UNAVAILABLE
assert not state.attributes.get(ATTR_ASSUMED_STATE)
async_fire_mqtt_message(hass, "availability_topic", "offline")
state = hass.states.get("fan.test")
assert state.state is STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "state-topic", "1")
state = hass.states.get("fan.test")
assert state.state is STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability_topic", "online")
state = hass.states.get("fan.test")
assert state.state is not STATE_UNAVAILABLE
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
await help_test_default_availability_payload(
hass, mqtt_mock, fan.DOMAIN, DEFAULT_CONFIG, True, "state-topic", "1"
)
async def test_custom_availability_payload(hass, mqtt_mock):
"""Test the availability payload."""
assert await async_setup_component(
hass,
fan.DOMAIN,
{
fan.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "state-topic",
"command_topic": "command-topic",
"availability_topic": "availability_topic",
"payload_available": "good",
"payload_not_available": "nogood",
}
},
"""Test availability by custom payload with defined topic."""
await help_test_custom_availability_payload(
hass, mqtt_mock, fan.DOMAIN, DEFAULT_CONFIG, True, "state-topic", "1"
)
state = hass.states.get("fan.test")
assert state.state is STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability_topic", "good")
state = hass.states.get("fan.test")
assert state.state is not STATE_UNAVAILABLE
assert not state.attributes.get(ATTR_ASSUMED_STATE)
async_fire_mqtt_message(hass, "availability_topic", "nogood")
state = hass.states.get("fan.test")
assert state.state is STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "state-topic", "1")
state = hass.states.get("fan.test")
assert state.state is STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability_topic", "good")
state = hass.states.get("fan.test")
assert state.state is not STATE_UNAVAILABLE
async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_via_mqtt_json_message(
hass, mqtt_mock, fan.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, fan.DOMAIN, DEFAULT_CONFIG
)
async def test_setting_attribute_with_template(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_with_template(
hass, mqtt_mock, fan.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_not_dict(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_not_dict(
hass, mqtt_mock, caplog, fan.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, fan.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_bad_JSON(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_bad_JSON(
hass, mqtt_mock, caplog, fan.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, fan.DOMAIN, DEFAULT_CONFIG
)
async def test_discovery_update_attr(hass, mqtt_mock, caplog):
"""Test update of discovered MQTTAttributes."""
data1 = (
'{ "name": "test",'
' "command_topic": "test_topic",'
' "json_attributes_topic": "attr-topic1" }'
)
data2 = (
'{ "name": "test",'
' "command_topic": "test_topic",'
' "json_attributes_topic": "attr-topic2" }'
)
await help_test_discovery_update_attr(
hass, mqtt_mock, caplog, fan.DOMAIN, data1, data2
hass, mqtt_mock, caplog, fan.DOMAIN, DEFAULT_CONFIG
)
@ -550,45 +468,34 @@ async def test_discovery_broken(hass, mqtt_mock, caplog):
await help_test_discovery_broken(hass, mqtt_mock, caplog, fan.DOMAIN, data1, data2)
async def test_entity_device_info_with_connection(hass, mqtt_mock):
"""Test MQTT fan device registry integration."""
await help_test_entity_device_info_with_connection(
hass, mqtt_mock, fan.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_with_identifier(hass, mqtt_mock):
"""Test MQTT fan device registry integration."""
await help_test_entity_device_info_with_identifier(
hass, mqtt_mock, fan.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, fan.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_update(hass, mqtt_mock):
"""Test device registry update."""
await help_test_entity_device_info_update(
hass, mqtt_mock, fan.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, fan.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_remove(hass, mqtt_mock):
"""Test device registry remove."""
config = {
"platform": "mqtt",
"name": "Test 1",
"state_topic": "test-topic",
"command_topic": "test-command-topic",
"device": {"identifiers": ["helloworld"]},
"unique_id": "veryunique",
}
await help_test_entity_device_info_remove(hass, mqtt_mock, fan.DOMAIN, config)
await help_test_entity_device_info_remove(
hass, mqtt_mock, fan.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_id_update(hass, mqtt_mock):
"""Test MQTT subscriptions are managed when entity_id is updated."""
config = {
fan.DOMAIN: [
{
"platform": "mqtt",
"name": "beer",
"state_topic": "test-topic",
"command_topic": "command-topic",
"availability_topic": "avty-topic",
"unique_id": "TOTALLY_UNIQUE",
}
]
}
await help_test_entity_id_update(hass, mqtt_mock, fan.DOMAIN, config)
await help_test_entity_id_update(hass, mqtt_mock, fan.DOMAIN, DEFAULT_CONFIG)

159
tests/components/mqtt/test_legacy_vacuum.py
View File

@ -16,25 +16,24 @@ from homeassistant.components.vacuum import (
ATTR_FAN_SPEED,
ATTR_STATUS,
)
from homeassistant.const import (
CONF_NAME,
CONF_PLATFORM,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.const import CONF_NAME, CONF_PLATFORM, STATE_OFF, STATE_ON
from homeassistant.setup import async_setup_component
from .common import (
help_test_availability_without_topic,
help_test_custom_availability_payload,
help_test_default_availability_payload,
help_test_discovery_broken,
help_test_discovery_removal,
help_test_discovery_update,
help_test_discovery_update_attr,
help_test_entity_device_info_remove,
help_test_entity_device_info_update,
help_test_entity_device_info_with_connection,
help_test_entity_device_info_with_identifier,
help_test_entity_id_update,
help_test_setting_attribute_via_mqtt_json_message,
help_test_setting_attribute_with_template,
help_test_unique_id,
help_test_update_with_json_attrs_bad_JSON,
help_test_update_with_json_attrs_not_dict,
@ -64,28 +63,7 @@ DEFAULT_CONFIG = {
mqttvacuum.CONF_FAN_SPEED_LIST: ["min", "medium", "high", "max"],
}
DEFAULT_CONFIG_ATTR = {
vacuum.DOMAIN: {
"platform": "mqtt",
"name": "test",
"json_attributes_topic": "attr-topic",
}
}
DEFAULT_CONFIG_DEVICE_INFO = {
"platform": "mqtt",
"name": "Test 1",
"command_topic": "test-command-topic",
"device": {
"identifiers": ["helloworld"],
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
},
"unique_id": "veryunique",
}
DEFAULT_CONFIG_2 = {vacuum.DOMAIN: {"platform": "mqtt", "name": "test"}}
async def test_default_supported_features(hass, mqtt_mock):
@ -499,89 +477,59 @@ async def test_missing_fan_speed_template(hass, mqtt_mock):
assert state is None
async def test_availability_without_topic(hass, mqtt_mock):
"""Test availability without defined availability topic."""
await help_test_availability_without_topic(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
config = deepcopy(DEFAULT_CONFIG)
config.update({"availability_topic": "availability-topic"})
assert await async_setup_component(hass, vacuum.DOMAIN, {vacuum.DOMAIN: config})
state = hass.states.get("vacuum.mqtttest")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "online")
state = hass.states.get("vacuum.mqtttest")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "offline")
state = hass.states.get("vacuum.mqtttest")
assert state.state == STATE_UNAVAILABLE
await help_test_default_availability_payload(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_custom_availability_payload(hass, mqtt_mock):
"""Test availability by custom payload with defined topic."""
config = deepcopy(DEFAULT_CONFIG)
config.update(
{
"availability_topic": "availability-topic",
"payload_available": "good",
"payload_not_available": "nogood",
}
await help_test_custom_availability_payload(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
assert await async_setup_component(hass, vacuum.DOMAIN, {vacuum.DOMAIN: config})
state = hass.states.get("vacuum.mqtttest")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "good")
state = hass.states.get("vacuum.mqtttest")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "nogood")
state = hass.states.get("vacuum.mqtttest")
assert state.state == STATE_UNAVAILABLE
async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_via_mqtt_json_message(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_setting_attribute_with_template(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_with_template(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_update_with_json_attrs_not_dict(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_not_dict(
hass, mqtt_mock, caplog, vacuum.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_update_with_json_attrs_bad_JSON(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_bad_JSON(
hass, mqtt_mock, caplog, vacuum.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_discovery_update_attr(hass, mqtt_mock, caplog):
"""Test update of discovered MQTTAttributes."""
data1 = (
'{ "name": "test",'
' "command_topic": "test_topic",'
' "json_attributes_topic": "attr-topic1" }'
)
data2 = (
'{ "name": "test",'
' "command_topic": "test_topic",'
' "json_attributes_topic": "attr-topic2" }'
)
await help_test_discovery_update_attr(
hass, mqtt_mock, caplog, vacuum.DOMAIN, data1, data2
hass, mqtt_mock, caplog, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
@ -608,7 +556,7 @@ async def test_unique_id(hass, mqtt_mock):
async def test_discovery_removal_vacuum(hass, mqtt_mock, caplog):
"""Test removal of discovered vacuum."""
data = json.dumps(DEFAULT_CONFIG_ATTR[vacuum.DOMAIN])
data = json.dumps(DEFAULT_CONFIG_2[vacuum.DOMAIN])
await help_test_discovery_removal(hass, mqtt_mock, caplog, vacuum.DOMAIN, data)
@ -630,45 +578,46 @@ async def test_discovery_broken(hass, mqtt_mock, caplog):
)
async def test_entity_device_info_with_connection(hass, mqtt_mock):
"""Test MQTT vacuum device registry integration."""
await help_test_entity_device_info_with_connection(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_entity_device_info_with_identifier(hass, mqtt_mock):
"""Test MQTT vacuum device registry integration."""
await help_test_entity_device_info_with_identifier(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_entity_device_info_update(hass, mqtt_mock):
"""Test device registry update."""
await help_test_entity_device_info_update(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_entity_device_info_remove(hass, mqtt_mock):
"""Test device registry remove."""
config = {
"platform": "mqtt",
"name": "Test 1",
"command_topic": "test-command-topic",
"device": {"identifiers": ["helloworld"]},
"unique_id": "veryunique",
}
await help_test_entity_device_info_remove(hass, mqtt_mock, vacuum.DOMAIN, config)
await help_test_entity_device_info_remove(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_entity_id_update(hass, mqtt_mock):
"""Test MQTT subscriptions are managed when entity_id is updated."""
config = {
vacuum.DOMAIN: [
{
"platform": "mqtt",
"name": "beer",
"battery_level_topic": "test-topic",
"battery_level_template": "{{ value_json.battery_level }}",
"command_topic": "command-topic",
"availability_topic": "avty-topic",
"unique_id": "TOTALLY_UNIQUE",
}
]
vacuum.DOMAIN: {
"platform": "mqtt",
"name": "test",
"battery_level_topic": "test-topic",
"battery_level_template": "{{ value_json.battery_level }}",
"command_topic": "command-topic",
"availability_topic": "avty-topic",
}
}
await help_test_entity_id_update(hass, mqtt_mock, vacuum.DOMAIN, config)
await help_test_entity_id_update(
hass, mqtt_mock, vacuum.DOMAIN, config, ["test-topic", "avty-topic"]
)

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

@ -158,25 +158,25 @@ from unittest.mock import patch
from homeassistant.components import light, mqtt
from homeassistant.components.mqtt.discovery import async_start
from homeassistant.const import (
ATTR_ASSUMED_STATE,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.const import ATTR_ASSUMED_STATE, STATE_OFF, STATE_ON
import homeassistant.core as ha
from homeassistant.setup import async_setup_component
from .common import (
help_test_availability_without_topic,
help_test_custom_availability_payload,
help_test_default_availability_payload,
help_test_discovery_broken,
help_test_discovery_removal,
help_test_discovery_update,
help_test_discovery_update_attr,
help_test_entity_device_info_remove,
help_test_entity_device_info_update,
help_test_entity_device_info_with_connection,
help_test_entity_device_info_with_identifier,
help_test_entity_id_update,
help_test_setting_attribute_via_mqtt_json_message,
help_test_setting_attribute_with_template,
help_test_unique_id,
help_test_update_with_json_attrs_bad_JSON,
help_test_update_with_json_attrs_not_dict,
@ -190,29 +190,8 @@ from tests.common import (
)
from tests.components.light import common
DEFAULT_CONFIG_ATTR = {
light.DOMAIN: {
"platform": "mqtt",
"name": "test",
"command_topic": "test-topic",
"json_attributes_topic": "attr-topic",
}
}
DEFAULT_CONFIG_DEVICE_INFO = {
"platform": "mqtt",
"name": "Test 1",
"state_topic": "test-topic",
"command_topic": "test-command-topic",
"device": {
"identifiers": ["helloworld"],
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
},
"unique_id": "veryunique",
DEFAULT_CONFIG = {
light.DOMAIN: {"platform": "mqtt", "name": "test", "command_topic": "test-topic"}
}
@ -1034,105 +1013,59 @@ async def test_on_command_rgb(hass, mqtt_mock):
mqtt_mock.async_publish.assert_called_once_with("test_light/set", "OFF", 0, False)
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
assert await async_setup_component(
hass,
light.DOMAIN,
{
light.DOMAIN: {
"platform": "mqtt",
"name": "test",
"command_topic": "test_light/set",
"brightness_command_topic": "test_light/bright",
"rgb_command_topic": "test_light/rgb",
"availability_topic": "availability-topic",
}
},
async def test_availability_without_topic(hass, mqtt_mock):
"""Test availability without defined availability topic."""
await help_test_availability_without_topic(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("light.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "online")
state = hass.states.get("light.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "offline")
state = hass.states.get("light.test")
assert state.state == STATE_UNAVAILABLE
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
await help_test_default_availability_payload(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_custom_availability_payload(hass, mqtt_mock):
"""Test availability by custom payload with defined topic."""
assert await async_setup_component(
hass,
light.DOMAIN,
{
light.DOMAIN: {
"platform": "mqtt",
"name": "test",
"command_topic": "test_light/set",
"brightness_command_topic": "test_light/bright",
"rgb_command_topic": "test_light/rgb",
"availability_topic": "availability-topic",
"payload_available": "good",
"payload_not_available": "nogood",
}
},
await help_test_custom_availability_payload(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("light.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "good")
state = hass.states.get("light.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "nogood")
state = hass.states.get("light.test")
assert state.state == STATE_UNAVAILABLE
async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_via_mqtt_json_message(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_setting_attribute_with_template(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_with_template(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_not_dict(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_not_dict(
hass, mqtt_mock, caplog, light.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, light.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_bad_JSON(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_bad_JSON(
hass, mqtt_mock, caplog, light.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, light.DOMAIN, DEFAULT_CONFIG
)
async def test_discovery_update_attr(hass, mqtt_mock, caplog):
"""Test update of discovered MQTTAttributes."""
data1 = (
'{ "name": "test",'
' "command_topic": "test_topic",'
' "json_attributes_topic": "attr-topic1" }'
)
data2 = (
'{ "name": "test",'
' "command_topic": "test_topic",'
' "json_attributes_topic": "attr-topic2" }'
)
await help_test_discovery_update_attr(
hass, mqtt_mock, caplog, light.DOMAIN, data1, data2
hass, mqtt_mock, caplog, light.DOMAIN, DEFAULT_CONFIG
)
@ -1213,45 +1146,34 @@ async def test_discovery_broken(hass, mqtt_mock, caplog):
)
async def test_entity_device_info_with_connection(hass, mqtt_mock):
"""Test MQTT light device registry integration."""
await help_test_entity_device_info_with_connection(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_with_identifier(hass, mqtt_mock):
"""Test MQTT light device registry integration."""
await help_test_entity_device_info_with_identifier(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_update(hass, mqtt_mock):
"""Test device registry update."""
await help_test_entity_device_info_update(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_remove(hass, mqtt_mock):
"""Test device registry remove."""
config = {
"platform": "mqtt",
"name": "Test 1",
"state_topic": "test-topic",
"command_topic": "test-command-topic",
"device": {"identifiers": ["helloworld"]},
"unique_id": "veryunique",
}
await help_test_entity_device_info_remove(hass, mqtt_mock, light.DOMAIN, config)
await help_test_entity_device_info_remove(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_id_update(hass, mqtt_mock):
"""Test MQTT subscriptions are managed when entity_id is updated."""
config = {
light.DOMAIN: [
{
"platform": "mqtt",
"name": "beer",
"state_topic": "test-topic",
"command_topic": "command-topic",
"availability_topic": "avty-topic",
"unique_id": "TOTALLY_UNIQUE",
}
]
}
await help_test_entity_id_update(hass, mqtt_mock, light.DOMAIN, config)
await help_test_entity_id_update(hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG)

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

@ -98,21 +98,25 @@ from homeassistant.const import (
ATTR_SUPPORTED_FEATURES,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
import homeassistant.core as ha
from homeassistant.setup import async_setup_component
from .common import (
help_test_availability_without_topic,
help_test_custom_availability_payload,
help_test_default_availability_payload,
help_test_discovery_broken,
help_test_discovery_removal,
help_test_discovery_update,
help_test_discovery_update_attr,
help_test_entity_device_info_remove,
help_test_entity_device_info_update,
help_test_entity_device_info_with_connection,
help_test_entity_device_info_with_identifier,
help_test_entity_id_update,
help_test_setting_attribute_via_mqtt_json_message,
help_test_setting_attribute_with_template,
help_test_unique_id,
help_test_update_with_json_attrs_bad_JSON,
help_test_update_with_json_attrs_not_dict,
@ -121,33 +125,15 @@ from .common import (
from tests.common import MockConfigEntry, async_fire_mqtt_message, mock_coro
from tests.components.light import common
DEFAULT_CONFIG_ATTR = {
DEFAULT_CONFIG = {
light.DOMAIN: {
"platform": "mqtt",
"schema": "json",
"name": "test",
"command_topic": "test-topic",
"json_attributes_topic": "attr-topic",
}
}
DEFAULT_CONFIG_DEVICE_INFO = {
"platform": "mqtt",
"name": "Test 1",
"schema": "json",
"state_topic": "test-topic",
"command_topic": "test-command-topic",
"device": {
"identifiers": ["helloworld"],
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
},
"unique_id": "veryunique",
}
class JsonValidator(object):
"""Helper to compare JSON."""
@ -883,107 +869,59 @@ async def test_invalid_color_brightness_and_white_values(hass, mqtt_mock):
assert state.attributes.get("white_value") == 255
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
assert await async_setup_component(
hass,
light.DOMAIN,
{
light.DOMAIN: {
"platform": "mqtt",
"schema": "json",
"name": "test",
"state_topic": "test_light_rgb",
"command_topic": "test_light_rgb/set",
"availability_topic": "availability-topic",
}
},
async def test_availability_without_topic(hass, mqtt_mock):
"""Test availability without defined availability topic."""
await help_test_availability_without_topic(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("light.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "online")
state = hass.states.get("light.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "offline")
state = hass.states.get("light.test")
assert state.state == STATE_UNAVAILABLE
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
await help_test_default_availability_payload(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_custom_availability_payload(hass, mqtt_mock):
"""Test availability by custom payload with defined topic."""
assert await async_setup_component(
hass,
light.DOMAIN,
{
light.DOMAIN: {
"platform": "mqtt",
"schema": "json",
"name": "test",
"state_topic": "test_light_rgb",
"command_topic": "test_light_rgb/set",
"availability_topic": "availability-topic",
"payload_available": "good",
"payload_not_available": "nogood",
}
},
await help_test_custom_availability_payload(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("light.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "good")
state = hass.states.get("light.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "nogood")
state = hass.states.get("light.test")
assert state.state == STATE_UNAVAILABLE
async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_via_mqtt_json_message(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_setting_attribute_with_template(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_with_template(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_not_dict(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_not_dict(
hass, mqtt_mock, caplog, light.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, light.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_bad_JSON(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_bad_JSON(
hass, mqtt_mock, caplog, light.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, light.DOMAIN, DEFAULT_CONFIG
)
async def test_discovery_update_attr(hass, mqtt_mock, caplog):
"""Test update of discovered MQTTAttributes."""
data1 = (
'{ "name": "test",'
' "schema": "json",'
' "command_topic": "test_topic",'
' "json_attributes_topic": "attr-topic1" }'
)
data2 = (
'{ "name": "test",'
' "schema": "json",'
' "command_topic": "test_topic",'
' "json_attributes_topic": "attr-topic2" }'
)
await help_test_discovery_update_attr(
hass, mqtt_mock, caplog, light.DOMAIN, data1, data2
hass, mqtt_mock, caplog, light.DOMAIN, DEFAULT_CONFIG
)
@ -1067,47 +1005,34 @@ async def test_discovery_broken(hass, mqtt_mock, caplog):
)
async def test_entity_device_info_with_connection(hass, mqtt_mock):
"""Test MQTT light device registry integration."""
await help_test_entity_device_info_with_connection(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_with_identifier(hass, mqtt_mock):
"""Test MQTT light device registry integration."""
await help_test_entity_device_info_with_identifier(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_update(hass, mqtt_mock):
"""Test device registry update."""
await help_test_entity_device_info_update(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_remove(hass, mqtt_mock):
"""Test device registry remove."""
config = {
"platform": "mqtt",
"name": "Test 1",
"schema": "json",
"state_topic": "test-topic",
"command_topic": "test-command-topic",
"device": {"identifiers": ["helloworld"]},
"unique_id": "veryunique",
}
await help_test_entity_device_info_remove(hass, mqtt_mock, light.DOMAIN, config)
await help_test_entity_device_info_remove(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_id_update(hass, mqtt_mock):
"""Test MQTT subscriptions are managed when entity_id is updated."""
config = {
light.DOMAIN: [
{
"platform": "mqtt",
"name": "beer",
"schema": "json",
"state_topic": "test-topic",
"command_topic": "command-topic",
"availability_topic": "avty-topic",
"unique_id": "TOTALLY_UNIQUE",
}
]
}
await help_test_entity_id_update(hass, mqtt_mock, light.DOMAIN, config)
await help_test_entity_id_update(hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG)

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

@ -30,25 +30,25 @@ from unittest.mock import patch
from homeassistant.components import light, mqtt
from homeassistant.components.mqtt.discovery import async_start
from homeassistant.const import (
ATTR_ASSUMED_STATE,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.const import ATTR_ASSUMED_STATE, STATE_OFF, STATE_ON
import homeassistant.core as ha
from homeassistant.setup import async_setup_component
from .common import (
help_test_availability_without_topic,
help_test_custom_availability_payload,
help_test_default_availability_payload,
help_test_discovery_broken,
help_test_discovery_removal,
help_test_discovery_update,
help_test_discovery_update_attr,
help_test_entity_device_info_remove,
help_test_entity_device_info_update,
help_test_entity_device_info_with_connection,
help_test_entity_device_info_with_identifier,
help_test_entity_id_update,
help_test_setting_attribute_via_mqtt_json_message,
help_test_setting_attribute_with_template,
help_test_unique_id,
help_test_update_with_json_attrs_bad_JSON,
help_test_update_with_json_attrs_not_dict,
@ -61,7 +61,7 @@ from tests.common import (
mock_coro,
)
DEFAULT_CONFIG_ATTR = {
DEFAULT_CONFIG = {
light.DOMAIN: {
"platform": "mqtt",
"schema": "template",
@ -69,29 +69,9 @@ DEFAULT_CONFIG_ATTR = {
"command_topic": "test-topic",
"command_on_template": "on,{{ transition }}",
"command_off_template": "off,{{ transition|d }}",
"json_attributes_topic": "attr-topic",
}
}
DEFAULT_CONFIG_DEVICE_INFO = {
"platform": "mqtt",
"name": "Test 1",
"schema": "template",
"state_topic": "test-topic",
"command_topic": "test-command-topic",
"command_on_template": "on,{{ transition }}",
"command_off_template": "off,{{ transition|d }}",
"device": {
"identifiers": ["helloworld"],
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
},
"unique_id": "veryunique",
}
async def test_setup_fails(hass, mqtt_mock):
"""Test that setup fails with missing required configuration items."""
@ -485,113 +465,59 @@ async def test_invalid_values(hass, mqtt_mock):
assert state.attributes.get("effect") == "rainbow"
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
assert await async_setup_component(
hass,
light.DOMAIN,
{
light.DOMAIN: {
"platform": "mqtt",
"schema": "template",
"name": "test",
"command_topic": "test_light_rgb/set",
"command_on_template": "on,{{ transition }}",
"command_off_template": "off,{{ transition|d }}",
"availability_topic": "availability-topic",
}
},
async def test_availability_without_topic(hass, mqtt_mock):
"""Test availability without defined availability topic."""
await help_test_availability_without_topic(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("light.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "online")
state = hass.states.get("light.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "offline")
state = hass.states.get("light.test")
assert state.state == STATE_UNAVAILABLE
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
await help_test_default_availability_payload(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_custom_availability_payload(hass, mqtt_mock):
"""Test availability by custom payload with defined topic."""
assert await async_setup_component(
hass,
light.DOMAIN,
{
light.DOMAIN: {
"platform": "mqtt",
"schema": "template",
"name": "test",
"command_topic": "test_light_rgb/set",
"command_on_template": "on,{{ transition }}",
"command_off_template": "off,{{ transition|d }}",
"availability_topic": "availability-topic",
"payload_available": "good",
"payload_not_available": "nogood",
}
},
await help_test_custom_availability_payload(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("light.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "good")
state = hass.states.get("light.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "nogood")
state = hass.states.get("light.test")
assert state.state == STATE_UNAVAILABLE
async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_via_mqtt_json_message(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_setting_attribute_with_template(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_with_template(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_not_dict(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_not_dict(
hass, mqtt_mock, caplog, light.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, light.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_bad_JSON(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_bad_JSON(
hass, mqtt_mock, caplog, light.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, light.DOMAIN, DEFAULT_CONFIG
)
async def test_discovery_update_attr(hass, mqtt_mock, caplog):
"""Test update of discovered MQTTAttributes."""
data1 = (
'{ "name": "test",'
' "schema": "template",'
' "command_topic": "test_topic",'
' "command_on_template": "on",'
' "command_off_template": "off",'
' "json_attributes_topic": "attr-topic1" }'
)
data2 = (
'{ "name": "test",'
' "schema": "template",'
' "command_topic": "test_topic",'
' "command_on_template": "on",'
' "command_off_template": "off",'
' "json_attributes_topic": "attr-topic2" }'
)
await help_test_discovery_update_attr(
hass, mqtt_mock, caplog, light.DOMAIN, data1, data2
hass, mqtt_mock, caplog, light.DOMAIN, DEFAULT_CONFIG
)
@ -691,48 +617,34 @@ async def test_discovery_broken(hass, mqtt_mock, caplog):
)
async def test_entity_device_info_with_connection(hass, mqtt_mock):
"""Test MQTT light device registry integration."""
await help_test_entity_device_info_with_connection(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_with_identifier(hass, mqtt_mock):
"""Test MQTT light device registry integration."""
await help_test_entity_device_info_with_identifier(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_update(hass, mqtt_mock):
"""Test device registry update."""
await help_test_entity_device_info_update(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_remove(hass, mqtt_mock):
"""Test device registry remove."""
config = {
"platform": "mqtt",
"name": "Test 1",
"state_topic": "test-topic",
"command_topic": "test-command-topic",
"device": {"identifiers": ["helloworld"]},
"unique_id": "veryunique",
}
await help_test_entity_device_info_remove(hass, mqtt_mock, light.DOMAIN, config)
await help_test_entity_device_info_remove(
hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_id_update(hass, mqtt_mock):
"""Test MQTT subscriptions are managed when entity_id is updated."""
config = {
light.DOMAIN: [
{
"platform": "mqtt",
"name": "beer",
"schema": "template",
"state_topic": "test-topic",
"command_topic": "command-topic",
"command_on_template": "on,{{ transition }}",
"command_off_template": "off,{{ transition|d }}",
"availability_topic": "avty-topic",
"unique_id": "TOTALLY_UNIQUE",
}
]
}
await help_test_entity_id_update(hass, mqtt_mock, light.DOMAIN, config)
await help_test_entity_id_update(hass, mqtt_mock, light.DOMAIN, DEFAULT_CONFIG)

168
tests/components/mqtt/test_lock.py
View File

@ -1,23 +1,23 @@
"""The tests for the MQTT lock platform."""
from homeassistant.components import lock
from homeassistant.const import (
ATTR_ASSUMED_STATE,
STATE_LOCKED,
STATE_UNAVAILABLE,
STATE_UNLOCKED,
)
from homeassistant.const import ATTR_ASSUMED_STATE, STATE_LOCKED, STATE_UNLOCKED
from homeassistant.setup import async_setup_component
from .common import (
help_test_availability_without_topic,
help_test_custom_availability_payload,
help_test_default_availability_payload,
help_test_discovery_broken,
help_test_discovery_removal,
help_test_discovery_update,
help_test_discovery_update_attr,
help_test_entity_device_info_remove,
help_test_entity_device_info_update,
help_test_entity_device_info_with_connection,
help_test_entity_device_info_with_identifier,
help_test_entity_id_update,
help_test_setting_attribute_via_mqtt_json_message,
help_test_setting_attribute_with_template,
help_test_unique_id,
help_test_update_with_json_attrs_bad_JSON,
help_test_update_with_json_attrs_not_dict,
@ -26,29 +26,8 @@ from .common import (
from tests.common import async_fire_mqtt_message
from tests.components.lock import common
DEFAULT_CONFIG_ATTR = {
lock.DOMAIN: {
"platform": "mqtt",
"name": "test",
"command_topic": "test-topic",
"json_attributes_topic": "attr-topic",
}
}
DEFAULT_CONFIG_DEVICE_INFO = {
"platform": "mqtt",
"name": "Test 1",
"state_topic": "test-topic",
"command_topic": "test-command-topic",
"device": {
"identifiers": ["helloworld"],
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
},
"unique_id": "veryunique",
DEFAULT_CONFIG = {
lock.DOMAIN: {"platform": "mqtt", "name": "test", "command_topic": "test-topic"}
}
@ -270,109 +249,59 @@ async def test_sending_mqtt_commands_and_explicit_optimistic(hass, mqtt_mock):
assert state.attributes.get(ATTR_ASSUMED_STATE)
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
assert await async_setup_component(
hass,
lock.DOMAIN,
{
lock.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "state-topic",
"command_topic": "command-topic",
"payload_lock": "LOCK",
"payload_unlock": "UNLOCK",
"availability_topic": "availability-topic",
}
},
async def test_availability_without_topic(hass, mqtt_mock):
"""Test availability without defined availability topic."""
await help_test_availability_without_topic(
hass, mqtt_mock, lock.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("lock.test")
assert state.state is STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "online")
state = hass.states.get("lock.test")
assert state.state is not STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "offline")
state = hass.states.get("lock.test")
assert state.state is STATE_UNAVAILABLE
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
await help_test_default_availability_payload(
hass, mqtt_mock, lock.DOMAIN, DEFAULT_CONFIG
)
async def test_custom_availability_payload(hass, mqtt_mock):
"""Test availability by custom payload with defined topic."""
assert await async_setup_component(
hass,
lock.DOMAIN,
{
lock.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "state-topic",
"command_topic": "command-topic",
"payload_lock": "LOCK",
"payload_unlock": "UNLOCK",
"state_locked": "LOCKED",
"state_unlocked": "UNLOCKED",
"availability_topic": "availability-topic",
"payload_available": "good",
"payload_not_available": "nogood",
}
},
await help_test_custom_availability_payload(
hass, mqtt_mock, lock.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("lock.test")
assert state.state is STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "good")
state = hass.states.get("lock.test")
assert state.state is not STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "nogood")
state = hass.states.get("lock.test")
assert state.state is STATE_UNAVAILABLE
async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_via_mqtt_json_message(
hass, mqtt_mock, lock.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, lock.DOMAIN, DEFAULT_CONFIG
)
async def test_setting_attribute_with_template(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_with_template(
hass, mqtt_mock, lock.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_not_dict(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_not_dict(
hass, mqtt_mock, caplog, lock.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, lock.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_bad_JSON(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_bad_JSON(
hass, mqtt_mock, caplog, lock.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, lock.DOMAIN, DEFAULT_CONFIG
)
async def test_discovery_update_attr(hass, mqtt_mock, caplog):
"""Test update of discovered MQTTAttributes."""
data1 = (
'{ "name": "test",'
' "command_topic": "test_topic",'
' "json_attributes_topic": "attr-topic1" }'
)
data2 = (
'{ "name": "test",'
' "command_topic": "test_topic",'
' "json_attributes_topic": "attr-topic2" }'
)
await help_test_discovery_update_attr(
hass, mqtt_mock, caplog, lock.DOMAIN, data1, data2
hass, mqtt_mock, caplog, lock.DOMAIN, DEFAULT_CONFIG
)
@ -429,45 +358,34 @@ async def test_discovery_broken(hass, mqtt_mock, caplog):
await help_test_discovery_broken(hass, mqtt_mock, caplog, lock.DOMAIN, data1, data2)
async def test_entity_device_info_with_connection(hass, mqtt_mock):
"""Test MQTT lock device registry integration."""
await help_test_entity_device_info_with_connection(
hass, mqtt_mock, lock.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_with_identifier(hass, mqtt_mock):
"""Test MQTT lock device registry integration."""
await help_test_entity_device_info_with_identifier(
hass, mqtt_mock, lock.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, lock.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_update(hass, mqtt_mock):
"""Test device registry update."""
await help_test_entity_device_info_update(
hass, mqtt_mock, lock.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, lock.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_remove(hass, mqtt_mock):
"""Test device registry remove."""
config = {
"platform": "mqtt",
"name": "Test 1",
"state_topic": "test-topic",
"command_topic": "test-command-topic",
"device": {"identifiers": ["helloworld"]},
"unique_id": "veryunique",
}
await help_test_entity_device_info_remove(hass, mqtt_mock, lock.DOMAIN, config)
await help_test_entity_device_info_remove(
hass, mqtt_mock, lock.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_id_update(hass, mqtt_mock):
"""Test MQTT subscriptions are managed when entity_id is updated."""
config = {
lock.DOMAIN: [
{
"platform": "mqtt",
"name": "beer",
"state_topic": "test-topic",
"command_topic": "command-topic",
"availability_topic": "avty-topic",
"unique_id": "TOTALLY_UNIQUE",
}
]
}
await help_test_entity_id_update(hass, mqtt_mock, lock.DOMAIN, config)
await help_test_entity_id_update(hass, mqtt_mock, lock.DOMAIN, DEFAULT_CONFIG)

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

@ -6,21 +6,26 @@ from unittest.mock import patch
from homeassistant.components import mqtt
from homeassistant.components.mqtt.discovery import async_start
import homeassistant.components.sensor as sensor
from homeassistant.const import EVENT_STATE_CHANGED, STATE_UNAVAILABLE
from homeassistant.const import EVENT_STATE_CHANGED
import homeassistant.core as ha
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from .common import (
help_test_availability_without_topic,
help_test_custom_availability_payload,
help_test_default_availability_payload,
help_test_discovery_broken,
help_test_discovery_removal,
help_test_discovery_update,
help_test_discovery_update_attr,
help_test_entity_device_info_remove,
help_test_entity_device_info_update,
help_test_entity_device_info_with_connection,
help_test_entity_device_info_with_identifier,
help_test_entity_id_update,
help_test_setting_attribute_via_mqtt_json_message,
help_test_setting_attribute_with_template,
help_test_unique_id,
help_test_update_with_json_attrs_bad_JSON,
help_test_update_with_json_attrs_not_dict,
@ -32,28 +37,8 @@ from tests.common import (
async_fire_time_changed,
)
DEFAULT_CONFIG_ATTR = {
sensor.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "test-topic",
"json_attributes_topic": "attr-topic",
}
}
DEFAULT_CONFIG_DEVICE_INFO = {
"platform": "mqtt",
"name": "Test 1",
"state_topic": "test-topic",
"device": {
"identifiers": ["helloworld"],
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
},
"unique_id": "veryunique",
DEFAULT_CONFIG = {
sensor.DOMAIN: {"platform": "mqtt", "name": "test", "state_topic": "test-topic"}
}
@ -234,65 +219,26 @@ async def test_force_update_enabled(hass, mqtt_mock):
assert len(events) == 2
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
assert await async_setup_component(
hass,
sensor.DOMAIN,
{
sensor.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "test-topic",
"availability_topic": "availability-topic",
}
},
async def test_availability_without_topic(hass, mqtt_mock):
"""Test availability without defined availability topic."""
await help_test_availability_without_topic(
hass, mqtt_mock, sensor.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("sensor.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "online")
state = hass.states.get("sensor.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "offline")
state = hass.states.get("sensor.test")
assert state.state == STATE_UNAVAILABLE
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
await help_test_default_availability_payload(
hass, mqtt_mock, sensor.DOMAIN, DEFAULT_CONFIG
)
async def test_custom_availability_payload(hass, mqtt_mock):
"""Test availability by custom payload with defined topic."""
assert await async_setup_component(
hass,
sensor.DOMAIN,
{
sensor.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "test-topic",
"availability_topic": "availability-topic",
"payload_available": "good",
"payload_not_available": "nogood",
}
},
await help_test_custom_availability_payload(
hass, mqtt_mock, sensor.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("sensor.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "good")
state = hass.states.get("sensor.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "nogood")
state = hass.states.get("sensor.test")
assert state.state == STATE_UNAVAILABLE
async def test_setting_sensor_attribute_via_legacy_mqtt_json_message(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
@ -432,72 +378,36 @@ async def test_valid_device_class(hass, mqtt_mock):
async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
config = {
sensor.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "test-topic",
"json_attributes_topic": "attr-topic",
}
}
await help_test_setting_attribute_via_mqtt_json_message(
hass, mqtt_mock, sensor.DOMAIN, config
hass, mqtt_mock, sensor.DOMAIN, DEFAULT_CONFIG
)
async def test_setting_attribute_with_template(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
assert await async_setup_component(
hass,
sensor.DOMAIN,
{
sensor.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "test-topic",
"json_attributes_topic": "attr-topic",
"json_attributes_template": "{{ value_json['Timer1'] | tojson }}",
}
},
await help_test_setting_attribute_with_template(
hass, mqtt_mock, sensor.DOMAIN, DEFAULT_CONFIG
)
async_fire_mqtt_message(
hass, "attr-topic", json.dumps({"Timer1": {"Arm": 0, "Time": "22:18"}})
)
state = hass.states.get("sensor.test")
assert state.attributes.get("Arm") == 0
assert state.attributes.get("Time") == "22:18"
async def test_update_with_json_attrs_not_dict(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_not_dict(
hass, mqtt_mock, caplog, sensor.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, sensor.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_bad_JSON(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_bad_JSON(
hass, mqtt_mock, caplog, sensor.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, sensor.DOMAIN, DEFAULT_CONFIG
)
async def test_discovery_update_attr(hass, mqtt_mock, caplog):
"""Test update of discovered MQTTAttributes."""
data1 = (
'{ "name": "test",'
' "state_topic": "test_topic",'
' "json_attributes_topic": "attr-topic1" }'
)
data2 = (
'{ "name": "test",'
' "state_topic": "test_topic",'
' "json_attributes_topic": "attr-topic2" }'
)
await help_test_discovery_update_attr(
hass, mqtt_mock, caplog, sensor.DOMAIN, data1, data2
hass, mqtt_mock, caplog, sensor.DOMAIN, DEFAULT_CONFIG
)
@ -546,46 +456,37 @@ async def test_discovery_broken(hass, mqtt_mock, caplog):
)
async def test_entity_device_info_with_connection(hass, mqtt_mock):
"""Test MQTT sensor device registry integration."""
await help_test_entity_device_info_with_connection(
hass, mqtt_mock, sensor.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_with_identifier(hass, mqtt_mock):
"""Test MQTT sensor device registry integration."""
await help_test_entity_device_info_with_identifier(
hass, mqtt_mock, sensor.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, sensor.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_update(hass, mqtt_mock):
"""Test device registry update."""
await help_test_entity_device_info_update(
hass, mqtt_mock, sensor.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, sensor.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_remove(hass, mqtt_mock):
"""Test device registry remove."""
config = {
"platform": "mqtt",
"name": "Test 1",
"state_topic": "test-topic",
"device": {"identifiers": ["helloworld"]},
"unique_id": "veryunique",
}
await help_test_entity_device_info_remove(hass, mqtt_mock, sensor.DOMAIN, config)
await help_test_entity_device_info_remove(
hass, mqtt_mock, sensor.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_id_update(hass, mqtt_mock):
"""Test MQTT subscriptions are managed when entity_id is updated."""
config = {
sensor.DOMAIN: [
{
"platform": "mqtt",
"name": "beer",
"state_topic": "test-topic",
"availability_topic": "avty-topic",
"unique_id": "TOTALLY_UNIQUE",
}
]
}
await help_test_entity_id_update(hass, mqtt_mock, sensor.DOMAIN, config)
await help_test_entity_id_update(hass, mqtt_mock, sensor.DOMAIN, DEFAULT_CONFIG)
async def test_entity_device_info_with_hub(hass, mqtt_mock):

148
tests/components/mqtt/test_state_vacuum.py
View File

@ -26,21 +26,25 @@ from homeassistant.const import (
CONF_NAME,
CONF_PLATFORM,
ENTITY_MATCH_ALL,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.setup import async_setup_component
from .common import (
help_test_availability_without_topic,
help_test_custom_availability_payload,
help_test_default_availability_payload,
help_test_discovery_broken,
help_test_discovery_removal,
help_test_discovery_update,
help_test_discovery_update_attr,
help_test_entity_device_info_remove,
help_test_entity_device_info_update,
help_test_entity_device_info_with_connection,
help_test_entity_device_info_with_identifier,
help_test_entity_id_update,
help_test_setting_attribute_via_mqtt_json_message,
help_test_setting_attribute_with_template,
help_test_unique_id,
help_test_update_with_json_attrs_bad_JSON,
help_test_update_with_json_attrs_not_dict,
@ -64,29 +68,8 @@ DEFAULT_CONFIG = {
mqttvacuum.CONF_FAN_SPEED_LIST: ["min", "medium", "high", "max"],
}
DEFAULT_CONFIG_ATTR = {
vacuum.DOMAIN: {
"platform": "mqtt",
"schema": "state",
"name": "test",
"json_attributes_topic": "attr-topic",
}
}
DEFAULT_CONFIG_DEVICE_INFO = {
"platform": "mqtt",
"schema": "state",
"name": "Test 1",
"command_topic": "test-command-topic",
"device": {
"identifiers": ["helloworld"],
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
},
"unique_id": "veryunique",
DEFAULT_CONFIG_2 = {
vacuum.DOMAIN: {"platform": "mqtt", "schema": "state", "name": "test"}
}
@ -327,91 +310,59 @@ async def test_status_invalid_json(hass, mqtt_mock):
assert state.state == STATE_UNKNOWN
async def test_availability_without_topic(hass, mqtt_mock):
"""Test availability without defined availability topic."""
await help_test_availability_without_topic(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
config = deepcopy(DEFAULT_CONFIG)
config.update({"availability_topic": "availability-topic"})
assert await async_setup_component(hass, vacuum.DOMAIN, {vacuum.DOMAIN: config})
state = hass.states.get("vacuum.mqtttest")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "online")
state = hass.states.get("vacuum.mqtttest")
assert STATE_UNAVAILABLE != state.state
async_fire_mqtt_message(hass, "availability-topic", "offline")
state = hass.states.get("vacuum.mqtttest")
assert state.state == STATE_UNAVAILABLE
await help_test_default_availability_payload(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_custom_availability_payload(hass, mqtt_mock):
"""Test availability by custom payload with defined topic."""
config = deepcopy(DEFAULT_CONFIG)
config.update(
{
"availability_topic": "availability-topic",
"payload_available": "good",
"payload_not_available": "nogood",
}
await help_test_custom_availability_payload(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
assert await async_setup_component(hass, vacuum.DOMAIN, {vacuum.DOMAIN: config})
state = hass.states.get("vacuum.mqtttest")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "good")
state = hass.states.get("vacuum.mqtttest")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability-topic", "nogood")
state = hass.states.get("vacuum.mqtttest")
assert state.state == STATE_UNAVAILABLE
async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_via_mqtt_json_message(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_setting_attribute_with_template(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_with_template(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_update_with_json_attrs_not_dict(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_not_dict(
hass, mqtt_mock, caplog, vacuum.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_update_with_json_attrs_bad_json(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_bad_JSON(
hass, mqtt_mock, caplog, vacuum.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_discovery_update_attr(hass, mqtt_mock, caplog):
"""Test update of discovered MQTTAttributes."""
data1 = (
'{ "name": "test",'
' "schema": "state",'
' "command_topic": "test_topic",'
' "json_attributes_topic": "attr-topic1" }'
)
data2 = (
'{ "name": "test",'
' "schema": "state",'
' "command_topic": "test_topic",'
' "json_attributes_topic": "attr-topic2" }'
)
await help_test_discovery_update_attr(
hass, mqtt_mock, caplog, vacuum.DOMAIN, data1, data2
hass, mqtt_mock, caplog, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
@ -462,47 +413,34 @@ async def test_discovery_broken(hass, mqtt_mock, caplog):
)
async def test_entity_device_info_with_connection(hass, mqtt_mock):
"""Test MQTT vacuum device registry integration."""
await help_test_entity_device_info_with_connection(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_entity_device_info_with_identifier(hass, mqtt_mock):
"""Test MQTT vacuum device registry integration."""
await help_test_entity_device_info_with_identifier(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_entity_device_info_update(hass, mqtt_mock):
"""Test device registry update."""
await help_test_entity_device_info_update(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_entity_device_info_remove(hass, mqtt_mock):
"""Test device registry remove."""
config = {
"platform": "mqtt",
"schema": "state",
"name": "Test 1",
"state_topic": "test-topic",
"command_topic": "test-command-topic",
"device": {"identifiers": ["helloworld"]},
"unique_id": "veryunique",
}
await help_test_entity_device_info_remove(hass, mqtt_mock, vacuum.DOMAIN, config)
await help_test_entity_device_info_remove(
hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2
)
async def test_entity_id_update(hass, mqtt_mock):
"""Test MQTT subscriptions are managed when entity_id is updated."""
config = {
vacuum.DOMAIN: [
{
"platform": "mqtt",
"schema": "state",
"name": "beer",
"state_topic": "test-topic",
"command_topic": "command-topic",
"availability_topic": "avty-topic",
"unique_id": "TOTALLY_UNIQUE",
}
]
}
await help_test_entity_id_update(hass, mqtt_mock, vacuum.DOMAIN, config)
await help_test_entity_id_update(hass, mqtt_mock, vacuum.DOMAIN, DEFAULT_CONFIG_2)

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

@ -3,25 +3,25 @@ from asynctest import patch
import pytest
from homeassistant.components import switch
from homeassistant.const import (
ATTR_ASSUMED_STATE,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.const import ATTR_ASSUMED_STATE, STATE_OFF, STATE_ON
import homeassistant.core as ha
from homeassistant.setup import async_setup_component
from .common import (
help_test_availability_without_topic,
help_test_custom_availability_payload,
help_test_default_availability_payload,
help_test_discovery_broken,
help_test_discovery_removal,
help_test_discovery_update,
help_test_discovery_update_attr,
help_test_entity_device_info_remove,
help_test_entity_device_info_update,
help_test_entity_device_info_with_connection,
help_test_entity_device_info_with_identifier,
help_test_entity_id_update,
help_test_setting_attribute_via_mqtt_json_message,
help_test_setting_attribute_with_template,
help_test_unique_id,
help_test_update_with_json_attrs_bad_JSON,
help_test_update_with_json_attrs_not_dict,
@ -30,29 +30,8 @@ from .common import (
from tests.common import async_fire_mqtt_message, async_mock_mqtt_component, mock_coro
from tests.components.switch import common
DEFAULT_CONFIG_ATTR = {
switch.DOMAIN: {
"platform": "mqtt",
"name": "test",
"command_topic": "test-topic",
"json_attributes_topic": "attr-topic",
}
}
DEFAULT_CONFIG_DEVICE_INFO = {
"platform": "mqtt",
"name": "Test 1",
"state_topic": "test-topic",
"command_topic": "test-command-topic",
"device": {
"identifiers": ["helloworld"],
"connections": [["mac", "02:5b:26:a8:dc:12"]],
"manufacturer": "Whatever",
"name": "Beer",
"model": "Glass",
"sw_version": "0.1-beta",
},
"unique_id": "veryunique",
DEFAULT_CONFIG = {
switch.DOMAIN: {"platform": "mqtt", "name": "test", "command_topic": "test-topic"}
}
@ -171,92 +150,47 @@ async def test_controlling_state_via_topic_and_json_message(hass, mock_publish):
assert state.state == STATE_OFF
async def test_default_availability_payload(hass, mock_publish):
"""Test the availability payload."""
assert await async_setup_component(
hass,
switch.DOMAIN,
{
switch.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "state-topic",
"command_topic": "command-topic",
"availability_topic": "availability_topic",
"payload_on": 1,
"payload_off": 0,
}
},
async def test_availability_without_topic(hass, mqtt_mock):
"""Test availability without defined availability topic."""
await help_test_availability_without_topic(
hass, mqtt_mock, switch.DOMAIN, DEFAULT_CONFIG
)
state = hass.states.get("switch.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability_topic", "online")
async def test_default_availability_payload(hass, mqtt_mock):
"""Test availability by default payload with defined topic."""
config = {
switch.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "state-topic",
"command_topic": "command-topic",
"payload_on": 1,
"payload_off": 0,
}
}
state = hass.states.get("switch.test")
assert state.state == STATE_OFF
assert not state.attributes.get(ATTR_ASSUMED_STATE)
async_fire_mqtt_message(hass, "availability_topic", "offline")
state = hass.states.get("switch.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "state-topic", "1")
state = hass.states.get("switch.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability_topic", "online")
state = hass.states.get("switch.test")
assert state.state == STATE_ON
async def test_custom_availability_payload(hass, mock_publish):
"""Test the availability payload."""
assert await async_setup_component(
hass,
switch.DOMAIN,
{
switch.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "state-topic",
"command_topic": "command-topic",
"availability_topic": "availability_topic",
"payload_on": 1,
"payload_off": 0,
"payload_available": "good",
"payload_not_available": "nogood",
}
},
await help_test_default_availability_payload(
hass, mqtt_mock, switch.DOMAIN, config, True, "state-topic", "1"
)
state = hass.states.get("switch.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability_topic", "good")
async def test_custom_availability_payload(hass, mqtt_mock):
"""Test availability by custom payload with defined topic."""
config = {
switch.DOMAIN: {
"platform": "mqtt",
"name": "test",
"state_topic": "state-topic",
"command_topic": "command-topic",
"payload_on": 1,
"payload_off": 0,
}
}
state = hass.states.get("switch.test")
assert state.state == STATE_OFF
assert not state.attributes.get(ATTR_ASSUMED_STATE)
async_fire_mqtt_message(hass, "availability_topic", "nogood")
state = hass.states.get("switch.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "state-topic", "1")
state = hass.states.get("switch.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, "availability_topic", "good")
state = hass.states.get("switch.test")
assert state.state == STATE_ON
await help_test_custom_availability_payload(
hass, mqtt_mock, switch.DOMAIN, config, True, "state-topic", "1"
)
async def test_custom_state_payload(hass, mock_publish):
@ -296,38 +230,35 @@ async def test_custom_state_payload(hass, mock_publish):
async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_via_mqtt_json_message(
hass, mqtt_mock, switch.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, switch.DOMAIN, DEFAULT_CONFIG
)
async def test_setting_attribute_with_template(hass, mqtt_mock):
"""Test the setting of attribute via MQTT with JSON payload."""
await help_test_setting_attribute_with_template(
hass, mqtt_mock, switch.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_not_dict(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_not_dict(
hass, mqtt_mock, caplog, switch.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, switch.DOMAIN, DEFAULT_CONFIG
)
async def test_update_with_json_attrs_bad_JSON(hass, mqtt_mock, caplog):
"""Test attributes get extracted from a JSON result."""
await help_test_update_with_json_attrs_bad_JSON(
hass, mqtt_mock, caplog, switch.DOMAIN, DEFAULT_CONFIG_ATTR
hass, mqtt_mock, caplog, switch.DOMAIN, DEFAULT_CONFIG
)
async def test_discovery_update_attr(hass, mqtt_mock, caplog):
"""Test update of discovered MQTTAttributes."""
data1 = (
'{ "name": "test",'
' "command_topic": "test_topic",'
' "json_attributes_topic": "attr-topic1" }'
)
data2 = (
'{ "name": "test",'
' "command_topic": "test_topic",'
' "json_attributes_topic": "attr-topic2" }'
)
await help_test_discovery_update_attr(
hass, mqtt_mock, caplog, switch.DOMAIN, data1, data2
hass, mqtt_mock, caplog, switch.DOMAIN, DEFAULT_CONFIG
)
@ -394,45 +325,34 @@ async def test_discovery_broken(hass, mqtt_mock, caplog):
)
async def test_entity_device_info_with_connection(hass, mqtt_mock):
"""Test MQTT switch device registry integration."""
await help_test_entity_device_info_with_connection(
hass, mqtt_mock, switch.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_with_identifier(hass, mqtt_mock):
"""Test MQTT switch device registry integration."""
await help_test_entity_device_info_with_identifier(
hass, mqtt_mock, switch.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, switch.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_update(hass, mqtt_mock):
"""Test device registry update."""
await help_test_entity_device_info_update(
hass, mqtt_mock, switch.DOMAIN, DEFAULT_CONFIG_DEVICE_INFO
hass, mqtt_mock, switch.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_device_info_remove(hass, mqtt_mock):
"""Test device registry remove."""
config = {
"platform": "mqtt",
"name": "Test 1",
"state_topic": "test-topic",
"command_topic": "test-command-topic",
"device": {"identifiers": ["helloworld"]},
"unique_id": "veryunique",
}
await help_test_entity_device_info_remove(hass, mqtt_mock, switch.DOMAIN, config)
await help_test_entity_device_info_remove(
hass, mqtt_mock, switch.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_id_update(hass, mqtt_mock):
"""Test MQTT subscriptions are managed when entity_id is updated."""
config = {
switch.DOMAIN: [
{
"platform": "mqtt",
"name": "beer",
"state_topic": "test-topic",
"command_topic": "command-topic",
"availability_topic": "avty-topic",
"unique_id": "TOTALLY_UNIQUE",
}
]
}
await help_test_entity_id_update(hass, mqtt_mock, switch.DOMAIN, config)
await help_test_entity_id_update(hass, mqtt_mock, switch.DOMAIN, DEFAULT_CONFIG)