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

Fixed mqtt subscription filter on sys $ topics (#8166)

Browse source 
* Fixed mqtt subscription filter on sys $ topics

* fixed linting issue

* added unit tests for $ topics and changed fix to use re.escape

* merge upstream/dev mqtt unit tests

* Update test_init.py
This commit is contained in:
natemason 2017-06-27 13:17:55 +08:00 • committed by Paulus Schoutsen
parent af54311718
commit 23400c4b0a
1 changed files with 34 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -249,6 +249,40 @@ class TestMQTT(unittest.TestCase):
self.hass.block_till_done()
self.assertEqual(0, len(self.calls))
def test_subscribe_topic_sys_root(self):
"""Test the subscription of $ root topics."""
mqtt.subscribe(self.hass, '$test-topic/subtree/on', self.record_calls)
fire_mqtt_message(self.hass, '$test-topic/subtree/on', 'test-payload')
self.hass.block_till_done()
self.assertEqual(1, len(self.calls))
self.assertEqual('$test-topic/subtree/on', self.calls[0][0])
self.assertEqual('test-payload', self.calls[0][1])
def test_subscribe_topic_sys_root_and_wildcard_topic(self):
"""Test the subscription of $ root and wildcard topics."""
mqtt.subscribe(self.hass, '$test-topic/#', self.record_calls)
fire_mqtt_message(self.hass, '$test-topic/some-topic', 'test-payload')
self.hass.block_till_done()
self.assertEqual(1, len(self.calls))
self.assertEqual('$test-topic/some-topic', self.calls[0][0])
self.assertEqual('test-payload', self.calls[0][1])
def test_subscribe_topic_sys_root_and_wildcard_subtree_topic(self):
"""Test the subscription of $ root and wildcard subtree topics."""
mqtt.subscribe(self.hass, '$test-topic/subtree/#', self.record_calls)
fire_mqtt_message(self.hass, '$test-topic/subtree/some-topic',
'test-payload')
self.hass.block_till_done()
self.assertEqual(1, len(self.calls))
self.assertEqual('$test-topic/subtree/some-topic', self.calls[0][0])
self.assertEqual('test-payload', self.calls[0][1])
def test_subscribe_special_characters(self):
"""Test the subscription to topics with special characters."""
topic = '/test-topic/$(.)[^]{-}'