mirror of
https://github.com/home-assistant/core.git
synced 2025-07-22 04:37:06 +00:00
Async tests for device tracker mqtt (#18680)
This commit is contained in:
Adam Mills
GitHub
parent
5e18d52302
commit
d24ea7da90
@ -1,147 +1,144 @@
|
||||
"""The tests for the MQTT device tracker platform."""
|
||||
import asyncio
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
import logging
|
||||
import os
|
||||
from asynctest import patch
|
||||
import pytest
|
||||
|
||||
from homeassistant.setup import setup_component
|
||||
from homeassistant.setup import async_setup_component
|
||||
from homeassistant.components import device_tracker
|
||||
from homeassistant.const import CONF_PLATFORM
|
||||
|
||||
from tests.common import (
|
||||
get_test_home_assistant, mock_mqtt_component, fire_mqtt_message)
|
||||
async_mock_mqtt_component, async_fire_mqtt_message)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TestComponentsDeviceTrackerMQTT(unittest.TestCase):
|
||||
"""Test MQTT device tracker platform."""
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_comp(hass):
|
||||
"""Initialize components."""
|
||||
hass.loop.run_until_complete(async_mock_mqtt_component(hass))
|
||||
yaml_devices = hass.config.path(device_tracker.YAML_DEVICES)
|
||||
yield
|
||||
if os.path.isfile(yaml_devices):
|
||||
os.remove(yaml_devices)
|
||||
|
||||
def setUp(self): # pylint: disable=invalid-name
|
||||
"""Set up things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
mock_mqtt_component(self.hass)
|
||||
|
||||
def tearDown(self): # pylint: disable=invalid-name
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
try:
|
||||
os.remove(self.hass.config.path(device_tracker.YAML_DEVICES))
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
async def test_ensure_device_tracker_platform_validation(hass):
|
||||
"""Test if platform validation was done."""
|
||||
async def mock_setup_scanner(hass, config, see, discovery_info=None):
|
||||
"""Check that Qos was added by validation."""
|
||||
assert 'qos' in config
|
||||
|
||||
def test_ensure_device_tracker_platform_validation(self):
|
||||
"""Test if platform validation was done."""
|
||||
@asyncio.coroutine
|
||||
def mock_setup_scanner(hass, config, see, discovery_info=None):
|
||||
"""Check that Qos was added by validation."""
|
||||
assert 'qos' in config
|
||||
with patch('homeassistant.components.device_tracker.mqtt.'
|
||||
'async_setup_scanner', autospec=True,
|
||||
side_effect=mock_setup_scanner) as mock_sp:
|
||||
|
||||
with patch('homeassistant.components.device_tracker.mqtt.'
|
||||
'async_setup_scanner', autospec=True,
|
||||
side_effect=mock_setup_scanner) as mock_sp:
|
||||
|
||||
dev_id = 'paulus'
|
||||
topic = '/location/paulus'
|
||||
assert setup_component(self.hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt',
|
||||
'devices': {dev_id: topic}
|
||||
}
|
||||
})
|
||||
assert mock_sp.call_count == 1
|
||||
|
||||
def test_new_message(self):
|
||||
"""Test new message."""
|
||||
dev_id = 'paulus'
|
||||
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
|
||||
topic = '/location/paulus'
|
||||
location = 'work'
|
||||
|
||||
self.hass.config.components = set(['mqtt', 'zone'])
|
||||
assert setup_component(self.hass, device_tracker.DOMAIN, {
|
||||
assert await async_setup_component(hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt',
|
||||
'devices': {dev_id: topic}
|
||||
}
|
||||
})
|
||||
fire_mqtt_message(self.hass, topic, location)
|
||||
self.hass.block_till_done()
|
||||
assert location == self.hass.states.get(entity_id).state
|
||||
assert mock_sp.call_count == 1
|
||||
|
||||
def test_single_level_wildcard_topic(self):
|
||||
"""Test single level wildcard topic."""
|
||||
dev_id = 'paulus'
|
||||
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
|
||||
subscription = '/location/+/paulus'
|
||||
topic = '/location/room/paulus'
|
||||
location = 'work'
|
||||
|
||||
self.hass.config.components = set(['mqtt', 'zone'])
|
||||
assert setup_component(self.hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt',
|
||||
'devices': {dev_id: subscription}
|
||||
}
|
||||
})
|
||||
fire_mqtt_message(self.hass, topic, location)
|
||||
self.hass.block_till_done()
|
||||
assert location == self.hass.states.get(entity_id).state
|
||||
async def test_new_message(hass):
|
||||
"""Test new message."""
|
||||
dev_id = 'paulus'
|
||||
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
|
||||
topic = '/location/paulus'
|
||||
location = 'work'
|
||||
|
||||
def test_multi_level_wildcard_topic(self):
|
||||
"""Test multi level wildcard topic."""
|
||||
dev_id = 'paulus'
|
||||
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
|
||||
subscription = '/location/#'
|
||||
topic = '/location/room/paulus'
|
||||
location = 'work'
|
||||
hass.config.components = set(['mqtt', 'zone'])
|
||||
assert await async_setup_component(hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt',
|
||||
'devices': {dev_id: topic}
|
||||
}
|
||||
})
|
||||
async_fire_mqtt_message(hass, topic, location)
|
||||
await hass.async_block_till_done()
|
||||
assert location == hass.states.get(entity_id).state
|
||||
|
||||
self.hass.config.components = set(['mqtt', 'zone'])
|
||||
assert setup_component(self.hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt',
|
||||
'devices': {dev_id: subscription}
|
||||
}
|
||||
})
|
||||
fire_mqtt_message(self.hass, topic, location)
|
||||
self.hass.block_till_done()
|
||||
assert location == self.hass.states.get(entity_id).state
|
||||
|
||||
def test_single_level_wildcard_topic_not_matching(self):
|
||||
"""Test not matching single level wildcard topic."""
|
||||
dev_id = 'paulus'
|
||||
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
|
||||
subscription = '/location/+/paulus'
|
||||
topic = '/location/paulus'
|
||||
location = 'work'
|
||||
async def test_single_level_wildcard_topic(hass):
|
||||
"""Test single level wildcard topic."""
|
||||
dev_id = 'paulus'
|
||||
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
|
||||
subscription = '/location/+/paulus'
|
||||
topic = '/location/room/paulus'
|
||||
location = 'work'
|
||||
|
||||
self.hass.config.components = set(['mqtt', 'zone'])
|
||||
assert setup_component(self.hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt',
|
||||
'devices': {dev_id: subscription}
|
||||
}
|
||||
})
|
||||
fire_mqtt_message(self.hass, topic, location)
|
||||
self.hass.block_till_done()
|
||||
assert self.hass.states.get(entity_id) is None
|
||||
hass.config.components = set(['mqtt', 'zone'])
|
||||
assert await async_setup_component(hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt',
|
||||
'devices': {dev_id: subscription}
|
||||
}
|
||||
})
|
||||
async_fire_mqtt_message(hass, topic, location)
|
||||
await hass.async_block_till_done()
|
||||
assert location == hass.states.get(entity_id).state
|
||||
|
||||
def test_multi_level_wildcard_topic_not_matching(self):
|
||||
"""Test not matching multi level wildcard topic."""
|
||||
dev_id = 'paulus'
|
||||
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
|
||||
subscription = '/location/#'
|
||||
topic = '/somewhere/room/paulus'
|
||||
location = 'work'
|
||||
|
||||
self.hass.config.components = set(['mqtt', 'zone'])
|
||||
assert setup_component(self.hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt',
|
||||
'devices': {dev_id: subscription}
|
||||
}
|
||||
})
|
||||
fire_mqtt_message(self.hass, topic, location)
|
||||
self.hass.block_till_done()
|
||||
assert self.hass.states.get(entity_id) is None
|
||||
async def test_multi_level_wildcard_topic(hass):
|
||||
"""Test multi level wildcard topic."""
|
||||
dev_id = 'paulus'
|
||||
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
|
||||
subscription = '/location/#'
|
||||
topic = '/location/room/paulus'
|
||||
location = 'work'
|
||||
|
||||
hass.config.components = set(['mqtt', 'zone'])
|
||||
assert await async_setup_component(hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt',
|
||||
'devices': {dev_id: subscription}
|
||||
}
|
||||
})
|
||||
async_fire_mqtt_message(hass, topic, location)
|
||||
await hass.async_block_till_done()
|
||||
assert location == hass.states.get(entity_id).state
|
||||
|
||||
|
||||
async def test_single_level_wildcard_topic_not_matching(hass):
|
||||
"""Test not matching single level wildcard topic."""
|
||||
dev_id = 'paulus'
|
||||
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
|
||||
subscription = '/location/+/paulus'
|
||||
topic = '/location/paulus'
|
||||
location = 'work'
|
||||
|
||||
hass.config.components = set(['mqtt', 'zone'])
|
||||
assert await async_setup_component(hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt',
|
||||
'devices': {dev_id: subscription}
|
||||
}
|
||||
})
|
||||
async_fire_mqtt_message(hass, topic, location)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id) is None
|
||||
|
||||
|
||||
async def test_multi_level_wildcard_topic_not_matching(hass):
|
||||
"""Test not matching multi level wildcard topic."""
|
||||
dev_id = 'paulus'
|
||||
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
|
||||
subscription = '/location/#'
|
||||
topic = '/somewhere/room/paulus'
|
||||
location = 'work'
|
||||
|
||||
hass.config.components = set(['mqtt', 'zone'])
|
||||
assert await async_setup_component(hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt',
|
||||
'devices': {dev_id: subscription}
|
||||
}
|
||||
})
|
||||
async_fire_mqtt_message(hass, topic, location)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id) is None
|
||||
|
||||
@ -1,17 +1,15 @@
|
||||
"""The tests for the JSON MQTT device tracker platform."""
|
||||
import asyncio
|
||||
import json
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
from asynctest import patch
|
||||
import logging
|
||||
import os
|
||||
import pytest
|
||||
|
||||
from homeassistant.setup import setup_component
|
||||
from homeassistant.setup import async_setup_component
|
||||
from homeassistant.components import device_tracker
|
||||
from homeassistant.const import CONF_PLATFORM
|
||||
|
||||
from tests.common import (
|
||||
get_test_home_assistant, mock_mqtt_component, fire_mqtt_message)
|
||||
from tests.common import async_mock_mqtt_component, async_fire_mqtt_message
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
@ -25,172 +23,172 @@ LOCATION_MESSAGE_INCOMPLETE = {
|
||||
'longitude': 2.0}
|
||||
|
||||
|
||||
class TestComponentsDeviceTrackerJSONMQTT(unittest.TestCase):
|
||||
"""Test JSON MQTT device tracker platform."""
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_comp(hass):
|
||||
"""Initialize components."""
|
||||
hass.loop.run_until_complete(async_mock_mqtt_component(hass))
|
||||
yaml_devices = hass.config.path(device_tracker.YAML_DEVICES)
|
||||
yield
|
||||
if os.path.isfile(yaml_devices):
|
||||
os.remove(yaml_devices)
|
||||
|
||||
def setUp(self): # pylint: disable=invalid-name
|
||||
"""Set up things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
mock_mqtt_component(self.hass)
|
||||
|
||||
def tearDown(self): # pylint: disable=invalid-name
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
try:
|
||||
os.remove(self.hass.config.path(device_tracker.YAML_DEVICES))
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
async def test_ensure_device_tracker_platform_validation(hass):
|
||||
"""Test if platform validation was done."""
|
||||
async def mock_setup_scanner(hass, config, see, discovery_info=None):
|
||||
"""Check that Qos was added by validation."""
|
||||
assert 'qos' in config
|
||||
|
||||
def test_ensure_device_tracker_platform_validation(self):
|
||||
"""Test if platform validation was done."""
|
||||
@asyncio.coroutine
|
||||
def mock_setup_scanner(hass, config, see, discovery_info=None):
|
||||
"""Check that Qos was added by validation."""
|
||||
assert 'qos' in config
|
||||
with patch('homeassistant.components.device_tracker.mqtt_json.'
|
||||
'async_setup_scanner', autospec=True,
|
||||
side_effect=mock_setup_scanner) as mock_sp:
|
||||
|
||||
with patch('homeassistant.components.device_tracker.mqtt_json.'
|
||||
'async_setup_scanner', autospec=True,
|
||||
side_effect=mock_setup_scanner) as mock_sp:
|
||||
|
||||
dev_id = 'paulus'
|
||||
topic = 'location/paulus'
|
||||
assert setup_component(self.hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt_json',
|
||||
'devices': {dev_id: topic}
|
||||
}
|
||||
})
|
||||
assert mock_sp.call_count == 1
|
||||
|
||||
def test_json_message(self):
|
||||
"""Test json location message."""
|
||||
dev_id = 'zanzito'
|
||||
topic = 'location/zanzito'
|
||||
location = json.dumps(LOCATION_MESSAGE)
|
||||
|
||||
assert setup_component(self.hass, device_tracker.DOMAIN, {
|
||||
dev_id = 'paulus'
|
||||
topic = 'location/paulus'
|
||||
assert await async_setup_component(hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt_json',
|
||||
'devices': {dev_id: topic}
|
||||
}
|
||||
})
|
||||
fire_mqtt_message(self.hass, topic, location)
|
||||
self.hass.block_till_done()
|
||||
state = self.hass.states.get('device_tracker.zanzito')
|
||||
assert state.attributes.get('latitude') == 2.0
|
||||
assert state.attributes.get('longitude') == 1.0
|
||||
assert mock_sp.call_count == 1
|
||||
|
||||
def test_non_json_message(self):
|
||||
"""Test receiving a non JSON message."""
|
||||
dev_id = 'zanzito'
|
||||
topic = 'location/zanzito'
|
||||
location = 'home'
|
||||
|
||||
assert setup_component(self.hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt_json',
|
||||
'devices': {dev_id: topic}
|
||||
}
|
||||
})
|
||||
async def test_json_message(hass):
|
||||
"""Test json location message."""
|
||||
dev_id = 'zanzito'
|
||||
topic = 'location/zanzito'
|
||||
location = json.dumps(LOCATION_MESSAGE)
|
||||
|
||||
with self.assertLogs(level='ERROR') as test_handle:
|
||||
fire_mqtt_message(self.hass, topic, location)
|
||||
self.hass.block_till_done()
|
||||
assert "ERROR:homeassistant.components.device_tracker.mqtt_json:" \
|
||||
"Error parsing JSON payload: home" in \
|
||||
test_handle.output[0]
|
||||
assert await async_setup_component(hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt_json',
|
||||
'devices': {dev_id: topic}
|
||||
}
|
||||
})
|
||||
async_fire_mqtt_message(hass, topic, location)
|
||||
await hass.async_block_till_done()
|
||||
state = hass.states.get('device_tracker.zanzito')
|
||||
assert state.attributes.get('latitude') == 2.0
|
||||
assert state.attributes.get('longitude') == 1.0
|
||||
|
||||
def test_incomplete_message(self):
|
||||
"""Test receiving an incomplete message."""
|
||||
dev_id = 'zanzito'
|
||||
topic = 'location/zanzito'
|
||||
location = json.dumps(LOCATION_MESSAGE_INCOMPLETE)
|
||||
|
||||
assert setup_component(self.hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt_json',
|
||||
'devices': {dev_id: topic}
|
||||
}
|
||||
})
|
||||
async def test_non_json_message(hass, caplog):
|
||||
"""Test receiving a non JSON message."""
|
||||
dev_id = 'zanzito'
|
||||
topic = 'location/zanzito'
|
||||
location = 'home'
|
||||
|
||||
with self.assertLogs(level='ERROR') as test_handle:
|
||||
fire_mqtt_message(self.hass, topic, location)
|
||||
self.hass.block_till_done()
|
||||
assert "ERROR:homeassistant.components.device_tracker.mqtt_json:" \
|
||||
"Skipping update for following data because of missing " \
|
||||
"or malformatted data: {\"longitude\": 2.0}" in \
|
||||
test_handle.output[0]
|
||||
assert await async_setup_component(hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt_json',
|
||||
'devices': {dev_id: topic}
|
||||
}
|
||||
})
|
||||
|
||||
def test_single_level_wildcard_topic(self):
|
||||
"""Test single level wildcard topic."""
|
||||
dev_id = 'zanzito'
|
||||
subscription = 'location/+/zanzito'
|
||||
topic = 'location/room/zanzito'
|
||||
location = json.dumps(LOCATION_MESSAGE)
|
||||
caplog.set_level(logging.ERROR)
|
||||
caplog.clear()
|
||||
async_fire_mqtt_message(hass, topic, location)
|
||||
await hass.async_block_till_done()
|
||||
assert "Error parsing JSON payload: home" in \
|
||||
caplog.text
|
||||
|
||||
assert setup_component(self.hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt_json',
|
||||
'devices': {dev_id: subscription}
|
||||
}
|
||||
})
|
||||
fire_mqtt_message(self.hass, topic, location)
|
||||
self.hass.block_till_done()
|
||||
state = self.hass.states.get('device_tracker.zanzito')
|
||||
assert state.attributes.get('latitude') == 2.0
|
||||
assert state.attributes.get('longitude') == 1.0
|
||||
|
||||
def test_multi_level_wildcard_topic(self):
|
||||
"""Test multi level wildcard topic."""
|
||||
dev_id = 'zanzito'
|
||||
subscription = 'location/#'
|
||||
topic = 'location/zanzito'
|
||||
location = json.dumps(LOCATION_MESSAGE)
|
||||
async def test_incomplete_message(hass, caplog):
|
||||
"""Test receiving an incomplete message."""
|
||||
dev_id = 'zanzito'
|
||||
topic = 'location/zanzito'
|
||||
location = json.dumps(LOCATION_MESSAGE_INCOMPLETE)
|
||||
|
||||
assert setup_component(self.hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt_json',
|
||||
'devices': {dev_id: subscription}
|
||||
}
|
||||
})
|
||||
fire_mqtt_message(self.hass, topic, location)
|
||||
self.hass.block_till_done()
|
||||
state = self.hass.states.get('device_tracker.zanzito')
|
||||
assert state.attributes.get('latitude') == 2.0
|
||||
assert state.attributes.get('longitude') == 1.0
|
||||
assert await async_setup_component(hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt_json',
|
||||
'devices': {dev_id: topic}
|
||||
}
|
||||
})
|
||||
|
||||
def test_single_level_wildcard_topic_not_matching(self):
|
||||
"""Test not matching single level wildcard topic."""
|
||||
dev_id = 'zanzito'
|
||||
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
|
||||
subscription = 'location/+/zanzito'
|
||||
topic = 'location/zanzito'
|
||||
location = json.dumps(LOCATION_MESSAGE)
|
||||
caplog.set_level(logging.ERROR)
|
||||
caplog.clear()
|
||||
async_fire_mqtt_message(hass, topic, location)
|
||||
await hass.async_block_till_done()
|
||||
assert "Skipping update for following data because of missing " \
|
||||
"or malformatted data: {\"longitude\": 2.0}" in \
|
||||
caplog.text
|
||||
|
||||
assert setup_component(self.hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt_json',
|
||||
'devices': {dev_id: subscription}
|
||||
}
|
||||
})
|
||||
fire_mqtt_message(self.hass, topic, location)
|
||||
self.hass.block_till_done()
|
||||
assert self.hass.states.get(entity_id) is None
|
||||
|
||||
def test_multi_level_wildcard_topic_not_matching(self):
|
||||
"""Test not matching multi level wildcard topic."""
|
||||
dev_id = 'zanzito'
|
||||
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
|
||||
subscription = 'location/#'
|
||||
topic = 'somewhere/zanzito'
|
||||
location = json.dumps(LOCATION_MESSAGE)
|
||||
async def test_single_level_wildcard_topic(hass):
|
||||
"""Test single level wildcard topic."""
|
||||
dev_id = 'zanzito'
|
||||
subscription = 'location/+/zanzito'
|
||||
topic = 'location/room/zanzito'
|
||||
location = json.dumps(LOCATION_MESSAGE)
|
||||
|
||||
assert setup_component(self.hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt_json',
|
||||
'devices': {dev_id: subscription}
|
||||
}
|
||||
})
|
||||
fire_mqtt_message(self.hass, topic, location)
|
||||
self.hass.block_till_done()
|
||||
assert self.hass.states.get(entity_id) is None
|
||||
assert await async_setup_component(hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt_json',
|
||||
'devices': {dev_id: subscription}
|
||||
}
|
||||
})
|
||||
async_fire_mqtt_message(hass, topic, location)
|
||||
await hass.async_block_till_done()
|
||||
state = hass.states.get('device_tracker.zanzito')
|
||||
assert state.attributes.get('latitude') == 2.0
|
||||
assert state.attributes.get('longitude') == 1.0
|
||||
|
||||
|
||||
async def test_multi_level_wildcard_topic(hass):
|
||||
"""Test multi level wildcard topic."""
|
||||
dev_id = 'zanzito'
|
||||
subscription = 'location/#'
|
||||
topic = 'location/zanzito'
|
||||
location = json.dumps(LOCATION_MESSAGE)
|
||||
|
||||
assert await async_setup_component(hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt_json',
|
||||
'devices': {dev_id: subscription}
|
||||
}
|
||||
})
|
||||
async_fire_mqtt_message(hass, topic, location)
|
||||
await hass.async_block_till_done()
|
||||
state = hass.states.get('device_tracker.zanzito')
|
||||
assert state.attributes.get('latitude') == 2.0
|
||||
assert state.attributes.get('longitude') == 1.0
|
||||
|
||||
|
||||
async def test_single_level_wildcard_topic_not_matching(hass):
|
||||
"""Test not matching single level wildcard topic."""
|
||||
dev_id = 'zanzito'
|
||||
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
|
||||
subscription = 'location/+/zanzito'
|
||||
topic = 'location/zanzito'
|
||||
location = json.dumps(LOCATION_MESSAGE)
|
||||
|
||||
assert await async_setup_component(hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt_json',
|
||||
'devices': {dev_id: subscription}
|
||||
}
|
||||
})
|
||||
async_fire_mqtt_message(hass, topic, location)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id) is None
|
||||
|
||||
|
||||
async def test_multi_level_wildcard_topic_not_matching(hass):
|
||||
"""Test not matching multi level wildcard topic."""
|
||||
dev_id = 'zanzito'
|
||||
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
|
||||
subscription = 'location/#'
|
||||
topic = 'somewhere/zanzito'
|
||||
location = json.dumps(LOCATION_MESSAGE)
|
||||
|
||||
assert await async_setup_component(hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {
|
||||
CONF_PLATFORM: 'mqtt_json',
|
||||
'devices': {dev_id: subscription}
|
||||
}
|
||||
})
|
||||
async_fire_mqtt_message(hass, topic, location)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id) is None
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user