evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Coerce ascii flags in MQTT publish service call.

Browse source 
Also fixes a bug where the test when no payload is given succeeds only

because we run the jinja templating out of memory. Sending an empty

payload instead of trying to fix it through validation for now.
This commit is contained in:
Jan Harkes 2016-04-03 03:54:05 -04:00 • committed by Paulus Schoutsen
parent 24b7a7b964
commit 9ce2081110
2 changed files with 25 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
homeassistant/components/mqtt/__init__.py
View file

@ -70,9 +70,10 @@ MQTT_PUBLISH_SCHEMA = vol.Schema({
vol.Required(ATTR_TOPIC): mqtt_topic, vol.Required(ATTR_TOPIC): mqtt_topic,
vol.Exclusive(ATTR_PAYLOAD, 'payload'): object, vol.Exclusive(ATTR_PAYLOAD, 'payload'): object,
vol.Exclusive(ATTR_PAYLOAD_TEMPLATE, 'payload'): cv.string, vol.Exclusive(ATTR_PAYLOAD_TEMPLATE, 'payload'): cv.string,
vol.Required(ATTR_QOS, default=DEFAULT_QOS): vol.In([0, 1, 2]), vol.Required(ATTR_QOS, default=DEFAULT_QOS):
vol.Required(ATTR_RETAIN, default=DEFAULT_RETAIN): vol.Coerce(bool), vol.All(vol.Coerce(int), vol.In([0, 1, 2])),
}) vol.Required(ATTR_RETAIN, default=DEFAULT_RETAIN): vol.Boolean(),
}, required=True)
def _build_publish_data(topic, qos, retain): def _build_publish_data(topic, qos, retain):
@ -194,7 +195,8 @@ def setup(hass, config):
qos = call.data[ATTR_QOS] qos = call.data[ATTR_QOS]
retain = call.data[ATTR_RETAIN] retain = call.data[ATTR_RETAIN]
try: try:
payload = payload or template.render(hass, payload_template) payload = (payload if payload_template is None else
template.render(hass, payload_template)) or ''
except template.jinja2.TemplateError as exc: except template.jinja2.TemplateError as exc:
_LOGGER.error( _LOGGER.error(
"Unable to publish to '%s': rendering payload template of " "Unable to publish to '%s': rendering payload template of "

21
tests/components/mqtt/test_init.py
View file

@ -104,14 +104,31 @@ class TestMQTT(unittest.TestCase):
def test_service_call_without_payload_or_payload_template(self): def test_service_call_without_payload_or_payload_template(self):
"""Test the service call without payload or payload template. """Test the service call without payload or payload template.
If neither 'payload' or 'payload_template' is provided then fail. Send empty message if neither 'payload' nor 'payload_template'
are provided.
""" """
# Call the service directly because the helper functions require you to # Call the service directly because the helper functions require you to
# provide a payload. # provide a payload.
self.hass.services.call(mqtt.DOMAIN, mqtt.SERVICE_PUBLISH, { self.hass.services.call(mqtt.DOMAIN, mqtt.SERVICE_PUBLISH, {
mqtt.ATTR_TOPIC: "test/topic" mqtt.ATTR_TOPIC: "test/topic"
}, blocking=True) }, blocking=True)
self.assertFalse(mqtt.MQTT_CLIENT.publish.called) self.assertTrue(mqtt.MQTT_CLIENT.publish.called)
self.assertEqual(mqtt.MQTT_CLIENT.publish.call_args[0][1], "")
def test_service_call_with_ascii_qos_retain_flags(self):
"""Test the service call with args that can be misinterpreted.
Empty payload message and ascii formatted qos and retain flags.
"""
self.hass.services.call(mqtt.DOMAIN, mqtt.SERVICE_PUBLISH, {
mqtt.ATTR_TOPIC: "test/topic",
mqtt.ATTR_PAYLOAD: "",
mqtt.ATTR_QOS: '2',
mqtt.ATTR_RETAIN: 'no'
}, blocking=True)
self.assertTrue(mqtt.MQTT_CLIENT.publish.called)
self.assertEqual(mqtt.MQTT_CLIENT.publish.call_args[0][2], 2)
self.assertFalse(mqtt.MQTT_CLIENT.publish.call_args[0][3])
def test_subscribe_topic(self): def test_subscribe_topic(self):
"""Test the subscription of a topic.""" """Test the subscription of a topic."""