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

Create additional mqtt helper function for using template payload.

Browse Source
This commit is contained in:
Flyte 2016-02-10 22:38:33 +00:00
parent d52e2019c0
commit 4e0c7f8a3d
2 changed files with 29 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -57,23 +57,28 @@ ATTR_RETAIN = 'retain'
MAX_RECONNECT_WAIT = 300 # seconds
# pylint: disable=too-many-arguments
def publish(hass, topic, payload=None, qos=None,
retain=None, payload_template=None):
"""Publish message to an MQTT topic."""
data = {
ATTR_TOPIC: topic,
ATTR_PAYLOAD: payload,
ATTR_PAYLOAD_TEMPLATE: payload_template
}
def _build_publish_data(topic, qos, retain):
"""Build the arguments for the publish service without the payload."""
data = {ATTR_TOPIC: topic}
if qos is not None:
data[ATTR_QOS] = qos
if retain is not None:
data[ATTR_RETAIN] = retain
return data
def publish(hass, topic, payload, qos=None, retain=None):
"""Publish message to an MQTT topic."""
data = _build_publish_data(topic, qos, retain)
data[ATTR_PAYLOAD] = payload
hass.services.call(DOMAIN, SERVICE_PUBLISH, data)
def publish_template(hass, topic, payload_template, qos=None, retain=None):
"""Publish message to an MQTT topic using a template payload."""
data = _build_publish_data(topic, qos, retain)
data[ATTR_PAYLOAD_TEMPLATE] = payload_template
hass.services.call(DOMAIN, SERVICE_PUBLISH, data)
# pylint: enable=too-many-arguments
def subscribe(hass, topic, callback, qos=DEFAULT_QOS):

20
tests/components/test_mqtt.py
View File

@ -80,8 +80,7 @@ class TestMQTT(unittest.TestCase):
"""
If 'payload_template' is provided and 'payload' is not, then render it.
"""
mqtt.publish(self.hass, "test/topic",
**{mqtt.ATTR_PAYLOAD_TEMPLATE: "{{ 1+1 }}"})
mqtt.publish_template(self.hass, "test/topic", "{{ 1+1 }}")
self.hass.pool.block_till_done()
self.assertTrue(mqtt.MQTT_CLIENT.publish.called)
self.assertEqual(mqtt.MQTT_CLIENT.publish.call_args[0][1], "2")
@ -92,9 +91,13 @@ class TestMQTT(unittest.TestCase):
"""
payload = "not a template"
payload_template = "a template"
mqtt.publish(self.hass, "test/topic", payload,
**{mqtt.ATTR_PAYLOAD_TEMPLATE: payload_template})
self.hass.pool.block_till_done()
# Call the service directly because the helper functions don't allow
# you to provide payload AND payload_template.
self.hass.services.call(mqtt.DOMAIN, mqtt.SERVICE_PUBLISH, {
mqtt.ATTR_TOPIC: "test/topic",
mqtt.ATTR_PAYLOAD: payload,
mqtt.ATTR_PAYLOAD_TEMPLATE: payload_template
}, blocking=True)
self.assertTrue(mqtt.MQTT_CLIENT.publish.called)
self.assertEqual(mqtt.MQTT_CLIENT.publish.call_args[0][1], payload)
@ -102,8 +105,11 @@ class TestMQTT(unittest.TestCase):
"""
If neither 'payload' or 'payload_template' is provided then fail.
"""
mqtt.publish(self.hass, "test/topic")
self.hass.pool.block_till_done()
# Call the service directly because the helper functions require you to
# provide a payload.
self.hass.services.call(mqtt.DOMAIN, mqtt.SERVICE_PUBLISH, {
mqtt.ATTR_TOPIC: "test/topic"
}, blocking=True)
self.assertFalse(mqtt.MQTT_CLIENT.publish.called)
def test_subscribe_topic(self):