Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add Tasmota integration (#39624) 

* Add Tasmota integration

* Refactor

* Add tests, small improvements

* isort

* Attempt to fix tests failing with Python 3.8

* Revert "Attempt to fix tests failing with Python 3.8"

This reverts commit 11454f8a00136f068ea27204183fa3e62f3cd263.

* Fix tests failing with Python 3.8

* Cleanup tests

* Address review comments

* Address review comments

* Address review comments

* Use MAC address for device identification

* Bump hatasmota

* Bump hatasmota

* Apply suggestions from code review

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Fix indentation

* Remove preparation for device remove WS API

* Address review comments

* Remove useless try-except

* Tweak

* Improve tests

* Tweak

* Address review comments

* Correct test

* Update manifest.json

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
pull/41361/head
Erik Montnemery 2020-10-06 14:51:58 +02:00 committed by GitHub
parent 3cf6535011
commit 06e9489fc7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
20 changed files with 1569 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CODEOWNERS
View File

@ -437,6 +437,7 @@ homeassistant/components/tado/* @michaelarnauts @bdraco
homeassistant/components/tag/* @balloob @dmulcahey
homeassistant/components/tahoma/* @philklei
homeassistant/components/tankerkoenig/* @guillempages
homeassistant/components/tasmota/* @emontnemery
homeassistant/components/tautulli/* @ludeeus
homeassistant/components/tellduslive/* @fredrike
homeassistant/components/template/* @PhracturedBlue @tetienne

30
homeassistant/components/mqtt/__init__.py
View File

@ -1465,3 +1465,33 @@ async def websocket_subscribe(hass, connection, msg):
)
connection.send_message(websocket_api.result_message(msg["id"]))
@callback
def async_subscribe_connection_status(hass, connection_status_callback):
"""Subscribe to MQTT connection changes."""
@callback
def connected():
hass.async_add_job(connection_status_callback, True)
@callback
def disconnected():
_LOGGER.error("Calling connection_status_callback, False")
hass.async_add_job(connection_status_callback, False)
subscriptions = {
"connect": async_dispatcher_connect(hass, MQTT_CONNECTED, connected),
"disconnect": async_dispatcher_connect(hass, MQTT_DISCONNECTED, disconnected),
}
def unsubscribe():
subscriptions["connect"]()
subscriptions["disconnect"]()
return unsubscribe
def is_connected(hass):
"""Return if MQTT client is connected."""
return hass.data[DATA_MQTT].connected

104
homeassistant/components/tasmota/__init__.py Normal file
View File

@ -0,0 +1,104 @@
"""The Tasmota integration."""
import logging
from hatasmota.const import (
CONF_MAC,
CONF_MANUFACTURER,
CONF_MODEL,
CONF_NAME,
CONF_SW_VERSION,
)
from hatasmota.discovery import clear_discovery_topic
from hatasmota.mqtt import TasmotaMQTTClient
from homeassistant.components import mqtt
from homeassistant.components.mqtt.subscription import (
async_subscribe_topics,
async_unsubscribe_topics,
)
from homeassistant.core import callback
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.typing import HomeAssistantType
from . import discovery
from .const import CONF_DISCOVERY_PREFIX
from .discovery import TASMOTA_DISCOVERY_DEVICE
_LOGGER = logging.getLogger(__name__)
DEVICE_MACS = "tasmota_devices"
async def async_setup(hass: HomeAssistantType, config: dict):
"""Set up the Tasmota component."""
return True
async def async_setup_entry(hass, entry):
"""Set up Tasmota from a config entry."""
hass.data[DEVICE_MACS] = {}
def _publish(*args, **kwds):
mqtt.async_publish(hass, *args, **kwds)
async def _subscribe_topics(sub_state, topics):
# Optionally mark message handlers as callback
for topic in topics.values():
if "msg_callback" in topic and "event_loop_safe" in topic:
topic["msg_callback"] = callback(topic["msg_callback"])
return await async_subscribe_topics(hass, sub_state, topics)
async def _unsubscribe_topics(sub_state):
return await async_unsubscribe_topics(hass, sub_state)
tasmota_mqtt = TasmotaMQTTClient(_publish, _subscribe_topics, _unsubscribe_topics)
discovery_prefix = entry.data[CONF_DISCOVERY_PREFIX]
await discovery.async_start(hass, discovery_prefix, entry, tasmota_mqtt)
async def async_discover_device(config, mac):
"""Discover and add a Tasmota device."""
await async_setup_device(hass, mac, config, entry, tasmota_mqtt)
async_dispatcher_connect(hass, TASMOTA_DISCOVERY_DEVICE, async_discover_device)
return True
async def _remove_device(hass, config_entry, mac, tasmota_mqtt):
"""Remove device from device registry."""
device_registry = await hass.helpers.device_registry.async_get_registry()
device = device_registry.async_get_device(set(), {(CONNECTION_NETWORK_MAC, mac)})
if device is None:
return
_LOGGER.debug("Removing tasmota device %s", mac)
device_registry.async_remove_device(device.id)
clear_discovery_topic(mac, config_entry.data[CONF_DISCOVERY_PREFIX], tasmota_mqtt)
async def _update_device(hass, config_entry, config):
"""Add or update device registry."""
device_registry = await hass.helpers.device_registry.async_get_registry()
config_entry_id = config_entry.entry_id
device_info = {
"connections": {(CONNECTION_NETWORK_MAC, config[CONF_MAC])},
"manufacturer": config[CONF_MANUFACTURER],
"model": config[CONF_MODEL],
"name": config[CONF_NAME],
"sw_version": config[CONF_SW_VERSION],
"config_entry_id": config_entry_id,
}
_LOGGER.debug("Adding or updating tasmota device %s", config[CONF_MAC])
device = device_registry.async_get_or_create(**device_info)
hass.data[DEVICE_MACS][device.id] = config[CONF_MAC]
async def async_setup_device(hass, mac, config, config_entry, tasmota_mqtt):
"""Set up the Tasmota device."""
if not config:
await _remove_device(hass, config_entry, mac, tasmota_mqtt)
else:
await _update_device(hass, config_entry, config)

56
homeassistant/components/tasmota/config_flow.py Normal file
View File

@ -0,0 +1,56 @@
"""Config flow for Tasmota."""
import logging
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.mqtt import valid_subscribe_topic
from .const import ( # pylint:disable=unused-import
CONF_DISCOVERY_PREFIX,
DEFAULT_PREFIX,
DOMAIN,
)
_LOGGER = logging.getLogger(__name__)
class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_PUSH
async def async_step_user(self, user_input=None):
"""Handle a flow initialized by the user."""
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
return await self.async_step_config()
async def async_step_config(self, user_input=None):
"""Confirm the setup."""
errors = {}
data = {CONF_DISCOVERY_PREFIX: DEFAULT_PREFIX}
if user_input is not None:
bad_prefix = False
if self.show_advanced_options:
prefix = user_input[CONF_DISCOVERY_PREFIX]
try:
valid_subscribe_topic(f"{prefix}/#")
except vol.Invalid:
errors["base"] = "invalid_discovery_topic"
bad_prefix = True
else:
data = user_input
if not bad_prefix:
return self.async_create_entry(title="Tasmota", data=data)
fields = {}
if self.show_advanced_options:
fields[vol.Optional(CONF_DISCOVERY_PREFIX, default=DEFAULT_PREFIX)] = str
return self.async_show_form(
step_id="config", data_schema=vol.Schema(fields), errors=errors
)

6
homeassistant/components/tasmota/const.py Normal file
View File

@ -0,0 +1,6 @@
"""Constants used by multiple Tasmota modules."""
CONF_DISCOVERY_PREFIX = "discovery_prefix"
DEFAULT_PREFIX = "tasmota/discovery"
DOMAIN = "tasmota"

121
homeassistant/components/tasmota/discovery.py Normal file
View File

@ -0,0 +1,121 @@
"""Support for MQTT discovery."""
import asyncio
import logging
from hatasmota.const import CONF_RELAY
from hatasmota.discovery import (
TasmotaDiscovery,
get_device_config as tasmota_get_device_config,
get_entities_for_platform as tasmota_get_entities_for_platform,
get_entity as tasmota_get_entity,
has_entities_with_platform as tasmota_has_entities_with_platform,
)
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.typing import HomeAssistantType
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
SUPPORTED_COMPONENTS = {
"switch": CONF_RELAY,
}
ALREADY_DISCOVERED = "tasmota_discovered_components"
CONFIG_ENTRY_IS_SETUP = "tasmota_config_entry_is_setup"
DATA_CONFIG_ENTRY_LOCK = "tasmota_config_entry_lock"
TASMOTA_DISCOVERY_DEVICE = "tasmota_discovery_device"
TASMOTA_DISCOVERY_ENTITY_NEW = "tasmota_discovery_entity_new_{}"
TASMOTA_DISCOVERY_ENTITY_UPDATED = "tasmota_discovery_entity_updated_{}_{}_{}"
def clear_discovery_hash(hass, discovery_hash):
"""Clear entry in ALREADY_DISCOVERED list."""
del hass.data[ALREADY_DISCOVERED][discovery_hash]
def set_discovery_hash(hass, discovery_hash):
"""Set entry in ALREADY_DISCOVERED list."""
hass.data[ALREADY_DISCOVERED][discovery_hash] = {}
async def async_start(
hass: HomeAssistantType, discovery_topic, config_entry, tasmota_mqtt
) -> bool:
"""Start MQTT Discovery."""
async def async_device_discovered(payload, mac):
"""Process the received message."""
if ALREADY_DISCOVERED not in hass.data:
hass.data[ALREADY_DISCOVERED] = {}
_LOGGER.debug("Received discovery data for tasmota device: %s", mac)
tasmota_device_config = tasmota_get_device_config(payload)
async_dispatcher_send(
hass, TASMOTA_DISCOVERY_DEVICE, tasmota_device_config, mac
)
async with hass.data[DATA_CONFIG_ENTRY_LOCK]:
for component, component_key in SUPPORTED_COMPONENTS.items():
if not tasmota_has_entities_with_platform(payload, component_key):
continue
config_entries_key = f"{component}.tasmota"
if config_entries_key not in hass.data[CONFIG_ENTRY_IS_SETUP]:
await hass.config_entries.async_forward_entry_setup(
config_entry, component
)
hass.data[CONFIG_ENTRY_IS_SETUP].add(config_entries_key)
for component, component_key in SUPPORTED_COMPONENTS.items():
tasmota_entities = tasmota_get_entities_for_platform(payload, component_key)
for (idx, tasmota_entity_config) in enumerate(tasmota_entities):
discovery_hash = (mac, component, idx)
if not tasmota_entity_config:
# Entity disabled, clean up entity registry
entity_registry = (
await hass.helpers.entity_registry.async_get_registry()
)
unique_id = "{}_{}_{}".format(*discovery_hash)
entity_id = entity_registry.async_get_entity_id(
component, DOMAIN, unique_id
)
if entity_id:
_LOGGER.debug(
"Removing entity: %s %s", component, discovery_hash
)
entity_registry.async_remove(entity_id)
continue
if discovery_hash in hass.data[ALREADY_DISCOVERED]:
_LOGGER.debug(
"Entity already added, sending update: %s %s",
component,
discovery_hash,
)
async_dispatcher_send(
hass,
TASMOTA_DISCOVERY_ENTITY_UPDATED.format(*discovery_hash),
tasmota_entity_config,
)
else:
_LOGGER.debug("Adding new entity: %s %s", component, discovery_hash)
hass.data[ALREADY_DISCOVERED][discovery_hash] = None
tasmota_entity = tasmota_get_entity(
tasmota_entity_config, component_key, tasmota_mqtt
)
async_dispatcher_send(
hass,
TASMOTA_DISCOVERY_ENTITY_NEW.format(component),
tasmota_entity,
discovery_hash,
)
hass.data[DATA_CONFIG_ENTRY_LOCK] = asyncio.Lock()
hass.data[CONFIG_ENTRY_IS_SETUP] = set()
tasmota_discovery = TasmotaDiscovery(discovery_topic, tasmota_mqtt)
await tasmota_discovery.start_discovery(async_device_discovered)

9
homeassistant/components/tasmota/manifest.json Normal file
View File

@ -0,0 +1,9 @@
{
"domain": "tasmota",
"name": "Tasmota (beta)",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/tasmota",
"requirements": ["hatasmota==0.0.9"],
"dependencies": ["mqtt"],
"codeowners": ["@emontnemery"]
}

111
homeassistant/components/tasmota/mixins.py Normal file
View File

@ -0,0 +1,111 @@
"""Tasnmota entity mixins."""
import logging
from homeassistant.components.mqtt import (
async_subscribe_connection_status,
is_connected as mqtt_connected,
)
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import Entity
from .discovery import (
TASMOTA_DISCOVERY_ENTITY_UPDATED,
clear_discovery_hash,
set_discovery_hash,
)
DATA_MQTT = "mqtt"
_LOGGER = logging.getLogger(__name__)
class TasmotaEntity(Entity):
"""Base class for Tasmota entities."""
def __init__(self, tasmota_entity) -> None:
"""Initialize."""
self._tasmota_entity = tasmota_entity
class TasmotaAvailability(TasmotaEntity):
"""Mixin used for platforms that report availability."""
def __init__(self, **kwds) -> None:
"""Initialize the availability mixin."""
self._available = False
super().__init__(**kwds)
async def async_added_to_hass(self) -> None:
"""Subscribe MQTT events."""
await super().async_added_to_hass()
self._tasmota_entity.set_on_availability_callback(self.availability_updated)
self.async_on_remove(
async_subscribe_connection_status(self.hass, self.async_mqtt_connected)
)
@callback
def availability_updated(self, available: bool) -> None:
"""Handle updated availability."""
self._available = available
self.async_write_ha_state()
@callback
def async_mqtt_connected(self, _):
"""Update state on connection/disconnection to MQTT broker."""
if not self.hass.is_stopping:
self.async_write_ha_state()
@property
def available(self) -> bool:
"""Return if the device is available."""
if not mqtt_connected(self.hass) and not self.hass.is_stopping:
return False
return self._available
class TasmotaDiscoveryUpdate(TasmotaEntity):
"""Mixin used to handle updated discovery message."""
def __init__(self, discovery_hash, discovery_update, **kwds) -> None:
"""Initialize the discovery update mixin."""
self._discovery_hash = discovery_hash
self._discovery_update = discovery_update
self._removed_from_hass = False
super().__init__(**kwds)
async def async_added_to_hass(self) -> None:
"""Subscribe to discovery updates."""
await super().async_added_to_hass()
self._removed_from_hass = False
async def discovery_callback(config):
"""Handle discovery update."""
_LOGGER.debug(
"Got update for entity with hash: %s '%s'",
self._discovery_hash,
config,
)
if not self._tasmota_entity.config_same(config):
# Changed payload: Notify component
_LOGGER.debug("Updating component: %s", self.entity_id)
await self._discovery_update(config)
else:
# Unchanged payload: Ignore to avoid changing states
_LOGGER.debug("Ignoring unchanged update for: %s", self.entity_id)
# Set in case the entity has been removed and is re-added, for example when changing entity_id
set_discovery_hash(self.hass, self._discovery_hash)
self.async_on_remove(
async_dispatcher_connect(
self.hass,
TASMOTA_DISCOVERY_ENTITY_UPDATED.format(*self._discovery_hash),
discovery_callback,
)
)
async def async_will_remove_from_hass(self) -> None:
"""Stop listening to signal and cleanup discovery data.."""
if not self._removed_from_hass:
clear_discovery_hash(self.hass, self._discovery_hash)
self._removed_from_hass = True

19
homeassistant/components/tasmota/strings.json Normal file
View File

@ -0,0 +1,19 @@
{
"config": {
"step": {
"config": {
"title": "Tasmota",
"description": "Please enter the Tasmota configuration.",
"data": {
"discovery_prefix": "Discovery topic prefix"
}
}
},
"abort": {
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]"
},
"error": {
"invalid_discovery_topic": "Invalid discovery topic prefix."
}
}
}

116
homeassistant/components/tasmota/switch.py Normal file
View File

@ -0,0 +1,116 @@
"""Support for Tasmota switches."""
import logging
from homeassistant.components import switch
from homeassistant.components.switch import SwitchEntity
from homeassistant.core import callback
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from .const import DOMAIN as TASMOTA_DOMAIN
from .discovery import TASMOTA_DISCOVERY_ENTITY_NEW
from .mixins import TasmotaAvailability, TasmotaDiscoveryUpdate
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up Tasmota switch dynamically through discovery."""
@callback
def async_discover(tasmota_entity, discovery_hash):
"""Discover and add a Tasmota switch."""
async_add_entities(
[
TasmotaSwitch(
tasmota_entity=tasmota_entity, discovery_hash=discovery_hash
)
]
)
async_dispatcher_connect(
hass,
TASMOTA_DISCOVERY_ENTITY_NEW.format(switch.DOMAIN, TASMOTA_DOMAIN),
async_discover,
)
class TasmotaSwitch(
TasmotaAvailability,
TasmotaDiscoveryUpdate,
SwitchEntity,
):
"""Representation of a Tasmota switch."""
def __init__(self, tasmota_entity, **kwds):
"""Initialize the Tasmota switch."""
self._state = False
self._sub_state = None
self._unique_id = tasmota_entity.unique_id
super().__init__(
discovery_update=self.discovery_update,
tasmota_entity=tasmota_entity,
**kwds,
)
async def async_added_to_hass(self):
"""Subscribe to MQTT events."""
await super().async_added_to_hass()
self._tasmota_entity.set_on_state_callback(self.state_updated)
await self._subscribe_topics()
async def discovery_update(self, update):
"""Handle updated discovery message."""
self._tasmota_entity.config_update(update)
await self._subscribe_topics()
self.async_write_ha_state()
@callback
def state_updated(self, state):
"""Handle new MQTT state messages."""
self._state = state
self.async_write_ha_state()
async def _subscribe_topics(self):
"""(Re)Subscribe to topics."""
await self._tasmota_entity.subscribe_topics()
async def async_will_remove_from_hass(self):
"""Unsubscribe when removed."""
await self._tasmota_entity.unsubscribe_topics()
await super().async_will_remove_from_hass()
@property
def should_poll(self):
"""Return the polling state."""
return False
@property
def name(self):
"""Return the name of the switch."""
return self._tasmota_entity.name
@property
def is_on(self):
"""Return true if device is on."""
return self._state
@property
def unique_id(self):
"""Return a unique ID."""
return self._tasmota_entity.unique_id
@property
def device_info(self):
"""Return a device description for device registry."""
return {"connections": {(CONNECTION_NETWORK_MAC, self._tasmota_entity.mac)}}
async def async_turn_on(self, **kwargs):
"""Turn the device on."""
self._tasmota_entity.set_state(True)
async def async_turn_off(self, **kwargs):
"""Turn the device off."""
self._tasmota_entity.set_state(False)

1
homeassistant/generated/config_flows.py
View File

@ -186,6 +186,7 @@ FLOWS = [
"syncthru",
"synology_dsm",
"tado",
"tasmota",
"tellduslive",
"tesla",
"tibber",

3
requirements_all.txt
View File

@ -728,6 +728,9 @@ hass-nabucasa==0.37.0
# homeassistant.components.splunk
hass_splunk==0.1.1
# homeassistant.components.tasmota
hatasmota==0.0.9
# homeassistant.components.jewish_calendar
hdate==0.9.5

3
requirements_test_all.txt
View File

@ -360,6 +360,9 @@ hangups==0.4.11
# homeassistant.components.cloud
hass-nabucasa==0.37.0
# homeassistant.components.tasmota
hatasmota==0.0.9
# homeassistant.components.jewish_calendar
hdate==0.9.5

1
tests/components/tasmota/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for Tasmota component."""

48
tests/components/tasmota/conftest.py Normal file
View File

@ -0,0 +1,48 @@
"""Test fixtures for Tasmota component."""
import pytest
from homeassistant import config_entries
from homeassistant.components.tasmota.const import (
CONF_DISCOVERY_PREFIX,
DEFAULT_PREFIX,
DOMAIN,
)
from tests.common import MockConfigEntry, mock_device_registry, mock_registry
@pytest.fixture
def device_reg(hass):
"""Return an empty, loaded, registry."""
return mock_device_registry(hass)
@pytest.fixture
def entity_reg(hass):
"""Return an empty, loaded, registry."""
return mock_registry(hass)
async def setup_tasmota_helper(hass):
"""Set up Tasmota."""
hass.config.components.add("tasmota")
entry = MockConfigEntry(
connection_class=config_entries.CONN_CLASS_LOCAL_PUSH,
data={CONF_DISCOVERY_PREFIX: DEFAULT_PREFIX},
domain=DOMAIN,
title="Tasmota",
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
assert "tasmota" in hass.config.components
@pytest.fixture
async def setup_tasmota(hass):
"""Set up Tasmota."""
await setup_tasmota_helper(hass)

328
tests/components/tasmota/test_common.py Normal file
View File

@ -0,0 +1,328 @@
"""Common test objects."""
import copy
import json
from hatasmota.const import (
CONF_MAC,
CONF_OFFLINE,
CONF_ONLINE,
CONF_PREFIX,
PREFIX_CMND,
PREFIX_TELE,
)
from hatasmota.utils import (
get_state_offline,
get_state_online,
get_topic_tele_state,
get_topic_tele_will,
)
from homeassistant.components.tasmota.const import DEFAULT_PREFIX
from homeassistant.const import STATE_UNAVAILABLE
from tests.async_mock import ANY
from tests.common import async_fire_mqtt_message
async def help_test_availability_when_connection_lost(
hass, mqtt_client_mock, mqtt_mock, domain, config
):
"""Test availability after MQTT disconnection."""
async_fire_mqtt_message(
hass,
f"{DEFAULT_PREFIX}/{config[CONF_MAC]}/config",
json.dumps(config),
)
await hass.async_block_till_done()
async_fire_mqtt_message(
hass,
get_topic_tele_will(config),
get_state_online(config),
)
state = hass.states.get(f"{domain}.test")
assert state.state != STATE_UNAVAILABLE
mqtt_mock.connected = False
await hass.async_add_executor_job(mqtt_client_mock.on_disconnect, None, None, 0)
await hass.async_block_till_done()
await hass.async_block_till_done()
await hass.async_block_till_done()
state = hass.states.get(f"{domain}.test")
assert state.state == STATE_UNAVAILABLE
mqtt_mock.connected = True
await hass.async_add_executor_job(mqtt_client_mock.on_connect, None, None, None, 0)
await hass.async_block_till_done()
await hass.async_block_till_done()
await hass.async_block_till_done()
state = hass.states.get(f"{domain}.test")
assert state.state != STATE_UNAVAILABLE
async def help_test_availability(
hass,
mqtt_mock,
domain,
config,
):
"""Test availability.
This is a test helper for the TasmotaAvailability mixin.
"""
async_fire_mqtt_message(
hass,
f"{DEFAULT_PREFIX}/{config[CONF_MAC]}/config",
json.dumps(config),
)
await hass.async_block_till_done()
state = hass.states.get(f"{domain}.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(
hass,
get_topic_tele_will(config),
get_state_online(config),
)
state = hass.states.get(f"{domain}.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(
hass,
get_topic_tele_will(config),
get_state_offline(config),
)
state = hass.states.get(f"{domain}.test")
assert state.state == STATE_UNAVAILABLE
async def help_test_availability_discovery_update(
hass,
mqtt_mock,
domain,
config,
):
"""Test update of discovered TasmotaAvailability.
This is a test helper for the TasmotaAvailability mixin.
"""
# customize availability topic
config1 = copy.deepcopy(config)
config1[CONF_PREFIX][PREFIX_TELE] = "tele1"
config1[CONF_OFFLINE] = "offline1"
config1[CONF_ONLINE] = "online1"
config2 = copy.deepcopy(config)
config2[CONF_PREFIX][PREFIX_TELE] = "tele2"
config2[CONF_OFFLINE] = "offline2"
config2[CONF_ONLINE] = "online2"
data1 = json.dumps(config1)
data2 = json.dumps(config2)
availability_topic1 = get_topic_tele_will(config1)
availability_topic2 = get_topic_tele_will(config2)
assert availability_topic1 != availability_topic2
offline1 = get_state_offline(config1)
offline2 = get_state_offline(config2)
assert offline1 != offline2
online1 = get_state_online(config1)
online2 = get_state_online(config2)
assert online1 != online2
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/{config1[CONF_MAC]}/config", data1)
await hass.async_block_till_done()
state = hass.states.get(f"{domain}.test")
assert state.state == STATE_UNAVAILABLE
async_fire_mqtt_message(hass, availability_topic1, online1)
state = hass.states.get(f"{domain}.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, availability_topic1, offline1)
state = hass.states.get(f"{domain}.test")
assert state.state == STATE_UNAVAILABLE
# Change availability settings
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/{config2[CONF_MAC]}/config", data2)
await hass.async_block_till_done()
# Verify we are no longer subscribing to the old topic or payload
async_fire_mqtt_message(hass, availability_topic1, online1)
async_fire_mqtt_message(hass, availability_topic1, online2)
async_fire_mqtt_message(hass, availability_topic2, online1)
state = hass.states.get(f"{domain}.test")
assert state.state == STATE_UNAVAILABLE
# Verify we are subscribing to the new topic
async_fire_mqtt_message(hass, availability_topic2, online2)
state = hass.states.get(f"{domain}.test")
assert state.state != STATE_UNAVAILABLE
async def help_test_discovery_removal(
hass, mqtt_mock, caplog, domain, config1, config2
):
"""Test removal of discovered entity."""
device_reg = await hass.helpers.device_registry.async_get_registry()
entity_reg = await hass.helpers.entity_registry.async_get_registry()
data1 = json.dumps(config1)
data2 = json.dumps(config2)
assert config1[CONF_MAC] == config2[CONF_MAC]
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/{config1[CONF_MAC]}/config", data1)
await hass.async_block_till_done()
# Verify device and entity registry entries are created
device_entry = device_reg.async_get_device(set(), {("mac", config1[CONF_MAC])})
assert device_entry is not None
entity_entry = entity_reg.async_get(f"{domain}.test")
assert entity_entry is not None
# Verify state is added
state = hass.states.get(f"{domain}.test")
assert state is not None
assert state.name == "Test"
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/{config2[CONF_MAC]}/config", data2)
await hass.async_block_till_done()
# Verify entity registry entries are cleared
device_entry = device_reg.async_get_device(set(), {("mac", config2[CONF_MAC])})
assert device_entry is not None
entity_entry = entity_reg.async_get(f"{domain}.test")
assert entity_entry is None
# Verify state is removed
state = hass.states.get(f"{domain}.test")
assert state is None
async def help_test_discovery_update_unchanged(
hass, mqtt_mock, caplog, domain, config, discovery_update
):
"""Test update of discovered component without changes.
This is a test helper for the MqttDiscoveryUpdate mixin.
"""
config1 = copy.deepcopy(config)
config2 = copy.deepcopy(config)
config2[CONF_PREFIX][PREFIX_CMND] = "cmnd2"
data1 = json.dumps(config1)
data2 = json.dumps(config2)
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/{config[CONF_MAC]}/config", data1)
await hass.async_block_till_done()
state = hass.states.get(f"{domain}.test")
assert state is not None
assert state.name == "Test"
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/{config[CONF_MAC]}/config", data1)
await hass.async_block_till_done()
assert not discovery_update.called
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/{config[CONF_MAC]}/config", data2)
await hass.async_block_till_done()
assert discovery_update.called
async def help_test_discovery_device_remove(hass, mqtt_mock, domain, config):
"""Test domain entity is removed when device is removed."""
device_reg = await hass.helpers.device_registry.async_get_registry()
entity_reg = await hass.helpers.entity_registry.async_get_registry()
config = copy.deepcopy(config)
unique_id = f"{config[CONF_MAC]}_{domain}_0"
data = json.dumps(config)
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/{config[CONF_MAC]}/config", data)
await hass.async_block_till_done()
device = device_reg.async_get_device(set(), {("mac", config[CONF_MAC])})
assert device is not None
assert entity_reg.async_get_entity_id(domain, "tasmota", unique_id)
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/{config[CONF_MAC]}/config", "")
await hass.async_block_till_done()
device = device_reg.async_get_device(set(), {("mac", config[CONF_MAC])})
assert device is None
assert not entity_reg.async_get_entity_id(domain, "tasmota", unique_id)
async def help_test_entity_id_update_subscriptions(
hass, mqtt_mock, domain, config, topics=None
):
"""Test MQTT subscriptions are managed when entity_id is updated."""
entity_reg = await hass.helpers.entity_registry.async_get_registry()
config = copy.deepcopy(config)
data = json.dumps(config)
mqtt_mock.async_subscribe.reset_mock()
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/{config[CONF_MAC]}/config", data)
await hass.async_block_till_done()
topics = [get_topic_tele_state(config), get_topic_tele_will(config)]
assert len(topics) > 0
state = hass.states.get(f"{domain}.test")
assert state is not None
assert mqtt_mock.async_subscribe.call_count == len(topics)
for topic in topics:
mqtt_mock.async_subscribe.assert_any_call(topic, ANY, ANY, ANY)
mqtt_mock.async_subscribe.reset_mock()
entity_reg.async_update_entity(f"{domain}.test", new_entity_id=f"{domain}.milk")
await hass.async_block_till_done()
state = hass.states.get(f"{domain}.test")
assert state is None
state = hass.states.get(f"{domain}.milk")
assert state is not None
for topic in topics:
mqtt_mock.async_subscribe.assert_any_call(topic, ANY, ANY, ANY)
async def help_test_entity_id_update_discovery_update(hass, mqtt_mock, domain, config):
"""Test MQTT discovery update after entity_id is updated."""
entity_reg = await hass.helpers.entity_registry.async_get_registry()
config = copy.deepcopy(config)
data = json.dumps(config)
topic = get_topic_tele_will(config)
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/{config[CONF_MAC]}/config", data)
await hass.async_block_till_done()
async_fire_mqtt_message(hass, topic, get_state_online(config))
state = hass.states.get(f"{domain}.test")
assert state.state != STATE_UNAVAILABLE
async_fire_mqtt_message(hass, topic, get_state_offline(config))
state = hass.states.get(f"{domain}.test")
assert state.state == STATE_UNAVAILABLE
entity_reg.async_update_entity(f"{domain}.test", new_entity_id=f"{domain}.milk")
await hass.async_block_till_done()
assert config[CONF_PREFIX][PREFIX_TELE] != "tele2"
config[CONF_PREFIX][PREFIX_TELE] = "tele2"
data = json.dumps(config)
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/{config[CONF_MAC]}/config", data)
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids(domain)) == 1
topic = get_topic_tele_will(config)
async_fire_mqtt_message(hass, topic, get_state_online(config))
state = hass.states.get(f"{domain}.milk")
assert state.state != STATE_UNAVAILABLE

61
tests/components/tasmota/test_config_flow.py Normal file
View File

@ -0,0 +1,61 @@
"""Test config flow."""
from tests.common import MockConfigEntry
async def test_user_setup(hass, mqtt_mock):
"""Test we can finish a config flow."""
result = await hass.config_entries.flow.async_init(
"tasmota", context={"source": "user"}
)
assert result["type"] == "form"
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] == "create_entry"
assert result["result"].data == {
"discovery_prefix": "tasmota/discovery",
}
async def test_user_setup_advanced(hass, mqtt_mock):
"""Test we can finish a config flow."""
result = await hass.config_entries.flow.async_init(
"tasmota", context={"source": "user", "show_advanced_options": True}
)
assert result["type"] == "form"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"discovery_prefix": "test_tasmota/discovery"}
)
assert result["type"] == "create_entry"
assert result["result"].data == {
"discovery_prefix": "test_tasmota/discovery",
}
async def test_user_setup_invalid_topic_prefix(hass, mqtt_mock):
"""Test if connection cannot be made."""
result = await hass.config_entries.flow.async_init(
"tasmota", context={"source": "user", "show_advanced_options": True}
)
assert result["type"] == "form"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"discovery_prefix": "tasmota/config/#"}
)
assert result["type"] == "form"
assert result["errors"]["base"] == "invalid_discovery_topic"
async def test_user_single_instance(hass, mqtt_mock):
"""Test we only allow a single config flow."""
MockConfigEntry(domain="tasmota").add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
"tasmota", context={"source": "user"}
)
assert result["type"] == "abort"
assert result["reason"] == "single_instance_allowed"

372
tests/components/tasmota/test_discovery.py Normal file
View File

@ -0,0 +1,372 @@
"""The tests for the MQTT discovery."""
import copy
import json
from hatasmota.const import CONF_ONLINE
import pytest
from homeassistant.components.tasmota.const import DEFAULT_PREFIX
from homeassistant.components.tasmota.discovery import ALREADY_DISCOVERED
from tests.async_mock import patch
from tests.common import async_fire_mqtt_message
from tests.components.tasmota.conftest import setup_tasmota_helper
DEFAULT_CONFIG = {
"dn": "My Device",
"fn": ["Beer", "Milk", "Three", "Four", "Five"],
"hn": "tasmota_49A3BC",
"mac": "00000049A3BC",
"md": "Sonoff 123",
"ofl": "offline",
CONF_ONLINE: "online",
"state": ["OFF", "ON", "TOGGLE", "HOLD"],
"sw": "2.3.3.4",
"t": "tasmota_49A3BC",
"t_f": "%topic%/%prefix%/",
"t_p": ["cmnd", "stat", "tele"],
"li": [0, 0, 0, 0, 0, 0, 0, 0],
"rl": [0, 0, 0, 0, 0, 0, 0, 0],
"se": [],
"ver": 1,
}
async def test_subscribing_config_topic(hass, mqtt_mock, setup_tasmota):
"""Test setting up discovery."""
discovery_topic = DEFAULT_PREFIX
assert mqtt_mock.async_subscribe.called
call_args = mqtt_mock.async_subscribe.mock_calls[0][1]
assert call_args[0] == discovery_topic + "/#"
assert call_args[2] == 0
async def test_valid_discovery_message(hass, mqtt_mock, caplog):
"""Test discovery callback called."""
config = copy.deepcopy(DEFAULT_CONFIG)
with patch(
"homeassistant.components.tasmota.discovery.tasmota_has_entities_with_platform"
) as mock_tasmota_has_entities:
await setup_tasmota_helper(hass)
async_fire_mqtt_message(
hass, f"{DEFAULT_PREFIX}/00000049A3BC/config", json.dumps(config)
)
await hass.async_block_till_done()
assert mock_tasmota_has_entities.called
async def test_invalid_topic(hass, mqtt_mock):
"""Test receiving discovery message on wrong topic."""
with patch(
"homeassistant.components.tasmota.discovery.tasmota_has_entities_with_platform"
) as mock_tasmota_has_entities:
await setup_tasmota_helper(hass)
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/123456/configuration", "{}")
await hass.async_block_till_done()
assert not mock_tasmota_has_entities.called
async def test_invalid_message(hass, mqtt_mock, caplog):
"""Test receiving an invalid message."""
with patch(
"homeassistant.components.tasmota.discovery.tasmota_has_entities_with_platform"
) as mock_tasmota_has_entities:
await setup_tasmota_helper(hass)
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/123456/config", "asd")
await hass.async_block_till_done()
assert "Invalid discovery message" in caplog.text
assert not mock_tasmota_has_entities.called
async def test_invalid_mac(hass, mqtt_mock, caplog):
"""Test topic is not matching device MAC."""
config = copy.deepcopy(DEFAULT_CONFIG)
with patch(
"homeassistant.components.tasmota.discovery.tasmota_has_entities_with_platform"
) as mock_tasmota_has_entities:
await setup_tasmota_helper(hass)
async_fire_mqtt_message(
hass, f"{DEFAULT_PREFIX}/00000049A3BA/config", json.dumps(config)
)
await hass.async_block_till_done()
assert "MAC mismatch" in caplog.text
assert not mock_tasmota_has_entities.called
async def test_correct_config_discovery(
hass, mqtt_mock, caplog, device_reg, entity_reg, setup_tasmota
):
"""Test receiving valid discovery message."""
config = copy.deepcopy(DEFAULT_CONFIG)
config["rl"][0] = 1
mac = config["mac"]
async_fire_mqtt_message(
hass,
f"{DEFAULT_PREFIX}/{mac}/config",
json.dumps(config),
)
await hass.async_block_till_done()
# Verify device and registry entries are created
device_entry = device_reg.async_get_device(set(), {("mac", mac)})
assert device_entry is not None
entity_entry = entity_reg.async_get("switch.beer")
assert entity_entry is not None
state = hass.states.get("switch.beer")
assert state is not None
assert state.name == "Beer"
assert (mac, "switch", 0) in hass.data[ALREADY_DISCOVERED]
async def test_device_discover(
hass, mqtt_mock, caplog, device_reg, entity_reg, setup_tasmota
):
"""Test setting up a device."""
config = copy.deepcopy(DEFAULT_CONFIG)
mac = config["mac"]
async_fire_mqtt_message(
hass,
f"{DEFAULT_PREFIX}/{mac}/config",
json.dumps(config),
)
await hass.async_block_till_done()
# Verify device and registry entries are created
device_entry = device_reg.async_get_device(set(), {("mac", mac)})
assert device_entry is not None
assert device_entry.manufacturer == "Tasmota"
assert device_entry.model == config["md"]
assert device_entry.name == config["dn"]
assert device_entry.sw_version == config["sw"]
async def test_device_update(
hass, mqtt_mock, caplog, device_reg, entity_reg, setup_tasmota
):
"""Test updating a device."""
config = copy.deepcopy(DEFAULT_CONFIG)
config["md"] = "Model 1"
config["dn"] = "Name 1"
config["sw"] = "v1.2.3.4"
mac = config["mac"]
async_fire_mqtt_message(
hass,
f"{DEFAULT_PREFIX}/{mac}/config",
json.dumps(config),
)
await hass.async_block_till_done()
# Verify device entry is created
device_entry = device_reg.async_get_device(set(), {("mac", mac)})
assert device_entry is not None
# Update device parameters
config["md"] = "Another model"
config["dn"] = "Another name"
config["sw"] = "v6.6.6"
async_fire_mqtt_message(
hass,
f"{DEFAULT_PREFIX}/{mac}/config",
json.dumps(config),
)
await hass.async_block_till_done()
# Verify device entry is updated
device_entry = device_reg.async_get_device(set(), {("mac", mac)})
assert device_entry is not None
assert device_entry.model == "Another model"
assert device_entry.name == "Another name"
assert device_entry.sw_version == "v6.6.6"
@pytest.mark.no_fail_on_log_exception
async def test_discovery_broken(hass, mqtt_mock, caplog, device_reg, setup_tasmota):
"""Test handling of exception when creating discovered device."""
config = copy.deepcopy(DEFAULT_CONFIG)
mac = config["mac"]
data = json.dumps(config)
# Trigger an exception when the entity is added
with patch(
"hatasmota.discovery.get_device_config_helper",
return_value=object(),
):
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/{mac}/config", data)
await hass.async_block_till_done()
# Verify device entry is not created
device_entry = device_reg.async_get_device(set(), {("mac", mac)})
assert device_entry is None
assert (
"Exception in async_discover_device when dispatching 'tasmota_discovery_device'"
in caplog.text
)
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/{mac}/config", data)
await hass.async_block_till_done()
# Verify device entry is created
device_entry = device_reg.async_get_device(set(), {("mac", mac)})
assert device_entry is not None
async def test_device_remove(
hass, mqtt_mock, caplog, device_reg, entity_reg, setup_tasmota
):
"""Test removing a discovered device."""
config = copy.deepcopy(DEFAULT_CONFIG)
mac = config["mac"]
async_fire_mqtt_message(
hass,
f"{DEFAULT_PREFIX}/{mac}/config",
json.dumps(config),
)
await hass.async_block_till_done()
# Verify device entry is created
device_entry = device_reg.async_get_device(set(), {("mac", mac)})
assert device_entry is not None
async_fire_mqtt_message(
hass,
f"{DEFAULT_PREFIX}/{mac}/config",
"",
)
await hass.async_block_till_done()
# Verify device entry is removed
device_entry = device_reg.async_get_device(set(), {("mac", mac)})
assert device_entry is None
async def test_device_remove_stale(hass, mqtt_mock, caplog, device_reg, setup_tasmota):
"""Test removing a stale (undiscovered) device does not throw."""
mac = "00000049A3BC"
config_entry = hass.config_entries.async_entries("tasmota")[0]
# Create a device
device_reg.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={("mac", mac)},
)
# Verify device entry was created
device_entry = device_reg.async_get_device(set(), {("mac", mac)})
assert device_entry is not None
# Remove the device
device_reg.async_remove_device(device_entry.id)
# Verify device entry is removed
device_entry = device_reg.async_get_device(set(), {("mac", mac)})
assert device_entry is None
async def test_device_rediscover(
hass, mqtt_mock, caplog, device_reg, entity_reg, setup_tasmota
):
"""Test removing a device."""
config = copy.deepcopy(DEFAULT_CONFIG)
mac = config["mac"]
async_fire_mqtt_message(
hass,
f"{DEFAULT_PREFIX}/{mac}/config",
json.dumps(config),
)
await hass.async_block_till_done()
# Verify device entry is created
device_entry1 = device_reg.async_get_device(set(), {("mac", mac)})
assert device_entry1 is not None
async_fire_mqtt_message(
hass,
f"{DEFAULT_PREFIX}/{mac}/config",
"",
)
await hass.async_block_till_done()
# Verify device entry is removed
device_entry = device_reg.async_get_device(set(), {("mac", mac)})
assert device_entry is None
async_fire_mqtt_message(
hass,
f"{DEFAULT_PREFIX}/{mac}/config",
json.dumps(config),
)
await hass.async_block_till_done()
# Verify device entry is created, and id is reused
device_entry = device_reg.async_get_device(set(), {("mac", mac)})
assert device_entry is not None
assert device_entry1.id == device_entry.id
async def test_entity_duplicate_discovery(hass, mqtt_mock, caplog, setup_tasmota):
"""Test entities are not duplicated."""
config = copy.deepcopy(DEFAULT_CONFIG)
config["rl"][0] = 1
mac = config["mac"]
async_fire_mqtt_message(
hass,
f"{DEFAULT_PREFIX}/{mac}/config",
json.dumps(config),
)
async_fire_mqtt_message(
hass,
f"{DEFAULT_PREFIX}/{mac}/config",
json.dumps(config),
)
await hass.async_block_till_done()
state = hass.states.get("switch.beer")
state_duplicate = hass.states.get("binary_sensor.beer1")
assert state is not None
assert state.name == "Beer"
assert state_duplicate is None
assert (
f"Entity already added, sending update: switch ('{mac}', 'switch', 0)"
in caplog.text
)
async def test_entity_duplicate_removal(hass, mqtt_mock, caplog, setup_tasmota):
"""Test removing entity twice."""
config = copy.deepcopy(DEFAULT_CONFIG)
config["rl"][0] = 1
mac = config["mac"]
async_fire_mqtt_message(
hass,
f"{DEFAULT_PREFIX}/{mac}/config",
json.dumps(config),
)
await hass.async_block_till_done()
config["rl"][0] = 0
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/{mac}/config", json.dumps(config))
await hass.async_block_till_done()
assert f"Removing entity: switch ('{mac}', 'switch', 0)" in caplog.text
caplog.clear()
async_fire_mqtt_message(hass, f"{DEFAULT_PREFIX}/{mac}/config", json.dumps(config))
await hass.async_block_till_done()
assert f"Removing entity: switch ('{mac}', 'switch', 0)" not in caplog.text

177
tests/components/tasmota/test_switch.py Normal file
View File

@ -0,0 +1,177 @@
"""The tests for the MQTT switch platform."""
import copy
import json
from hatasmota.const import CONF_ONLINE
from homeassistant.components import switch
from homeassistant.components.tasmota.const import DEFAULT_PREFIX
from homeassistant.const import ATTR_ASSUMED_STATE, STATE_OFF, STATE_ON
from .test_common import (
help_test_availability,
help_test_availability_discovery_update,
help_test_availability_when_connection_lost,
help_test_discovery_device_remove,
help_test_discovery_removal,
help_test_discovery_update_unchanged,
help_test_entity_id_update_discovery_update,
help_test_entity_id_update_subscriptions,
)
from tests.async_mock import patch
from tests.common import async_fire_mqtt_message
from tests.components.switch import common
DEFAULT_CONFIG = {
"dn": "My Device",
"fn": ["Test", "Beer", "Milk", "Four", "Five"],
"hn": "tasmota_49A3BC",
"mac": "00000049A3BC",
"md": "Sonoff 123",
"ofl": "offline",
CONF_ONLINE: "online",
"state": ["OFF", "ON", "TOGGLE", "HOLD"],
"sw": "2.3.3.4",
"t": "tasmota_49A3BC",
"t_f": "%topic%/%prefix%/",
"t_p": ["cmnd", "stat", "tele"],
"li": [0, 0, 0, 0, 0, 0, 0, 0],
"rl": [1, 0, 0, 0, 0, 0, 0, 0],
"se": [],
"ver": 1,
}
async def test_controlling_state_via_mqtt(hass, mqtt_mock, setup_tasmota):
"""Test state update via MQTT."""
config = copy.deepcopy(DEFAULT_CONFIG)
mac = config["mac"]
async_fire_mqtt_message(
hass,
f"{DEFAULT_PREFIX}/{mac}/config",
json.dumps(config),
)
await hass.async_block_till_done()
state = hass.states.get("switch.test")
assert state.state == "unavailable"
assert not state.attributes.get(ATTR_ASSUMED_STATE)
async_fire_mqtt_message(hass, "tasmota_49A3BC/tele/LWT", "online")
state = hass.states.get("switch.test")
assert state.state == STATE_OFF
assert not state.attributes.get(ATTR_ASSUMED_STATE)
async_fire_mqtt_message(hass, "tasmota_49A3BC/tele/STATE", '{"POWER":"ON"}')
state = hass.states.get("switch.test")
assert state.state == STATE_ON
async_fire_mqtt_message(hass, "tasmota_49A3BC/tele/STATE", '{"POWER":"OFF"}')
state = hass.states.get("switch.test")
assert state.state == STATE_OFF
async def test_sending_mqtt_commands(hass, mqtt_mock, setup_tasmota):
"""Test the sending MQTT commands."""
config = copy.deepcopy(DEFAULT_CONFIG)
mac = config["mac"]
async_fire_mqtt_message(
hass,
f"{DEFAULT_PREFIX}/{mac}/config",
json.dumps(config),
)
await hass.async_block_till_done()
async_fire_mqtt_message(hass, "tasmota_49A3BC/tele/LWT", "online")
state = hass.states.get("switch.test")
assert state.state == STATE_OFF
# Turn the switch on and verify MQTT message is sent
await common.async_turn_on(hass, "switch.test")
mqtt_mock.async_publish.assert_called_once_with(
"tasmota_49A3BC/cmnd/POWER1", "ON", 0, False
)
mqtt_mock.async_publish.reset_mock()
# Tasmota is not optimistic, the state should still be off
state = hass.states.get("switch.test")
assert state.state == STATE_OFF
# Turn the switch off and verify MQTT message is sent
await common.async_turn_off(hass, "switch.test")
mqtt_mock.async_publish.assert_called_once_with(
"tasmota_49A3BC/cmnd/POWER1", "OFF", 0, False
)
state = hass.states.get("switch.test")
assert state.state == STATE_OFF
async def test_availability_when_connection_lost(
hass, mqtt_client_mock, mqtt_mock, setup_tasmota
):
"""Test availability after MQTT disconnection."""
await help_test_availability_when_connection_lost(
hass, mqtt_client_mock, mqtt_mock, switch.DOMAIN, DEFAULT_CONFIG
)
async def test_availability(hass, mqtt_mock, setup_tasmota):
"""Test availability."""
await help_test_availability(hass, mqtt_mock, switch.DOMAIN, DEFAULT_CONFIG)
async def test_availability_discovery_update(hass, mqtt_mock, setup_tasmota):
"""Test availability discovery update."""
await help_test_availability_discovery_update(
hass, mqtt_mock, switch.DOMAIN, DEFAULT_CONFIG
)
async def test_discovery_removal_switch(hass, mqtt_mock, caplog, setup_tasmota):
"""Test removal of discovered switch."""
config1 = copy.deepcopy(DEFAULT_CONFIG)
config2 = copy.deepcopy(DEFAULT_CONFIG)
config2["rl"][0] = 0
await help_test_discovery_removal(
hass, mqtt_mock, caplog, switch.DOMAIN, config1, config2
)
async def test_discovery_update_unchanged_switch(
hass, mqtt_mock, caplog, setup_tasmota
):
"""Test update of discovered switch."""
with patch(
"homeassistant.components.tasmota.switch.TasmotaSwitch.discovery_update"
) as discovery_update:
await help_test_discovery_update_unchanged(
hass, mqtt_mock, caplog, switch.DOMAIN, DEFAULT_CONFIG, discovery_update
)
async def test_discovery_device_remove(hass, mqtt_mock, setup_tasmota):
"""Test device registry remove."""
await help_test_discovery_device_remove(
hass, mqtt_mock, switch.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_id_update_subscriptions(hass, mqtt_mock, setup_tasmota):
"""Test MQTT subscriptions are managed when entity_id is updated."""
await help_test_entity_id_update_subscriptions(
hass, mqtt_mock, switch.DOMAIN, DEFAULT_CONFIG
)
async def test_entity_id_update_discovery_update(hass, mqtt_mock, setup_tasmota):
"""Test MQTT discovery update when entity_id is updated."""
await help_test_entity_id_update_discovery_update(
hass, mqtt_mock, switch.DOMAIN, DEFAULT_CONFIG
)

2
tests/conftest.py
View File

@ -357,10 +357,12 @@ def mqtt_client_mock(hass):
return FakeInfo(mid)
def _subscribe(topic, qos=0):
mid = get_mid()
mock_client.on_subscribe(0, 0, mid)
return (0, mid)
def _unsubscribe(topic):
mid = get_mid()
mock_client.on_unsubscribe(0, 0, mid)
return (0, mid)