forked from home-assistant/core
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Async tests for device tracker mqtt (home-assistant#18680)
- Loading branch information
Showing
2 changed files
with
287 additions
and
292 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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.""" | ||
|
|
||
| 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 | ||
|
|
||
| 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: | ||
|
|
||
| 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, { | ||
| 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 | ||
|
|
||
| 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' | ||
| @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) | ||
|
|
||
| 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_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' | ||
| 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 | ||
|
|
||
| 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 | ||
| with patch('homeassistant.components.device_tracker.mqtt.' | ||
| 'async_setup_scanner', autospec=True, | ||
| side_effect=mock_setup_scanner) as mock_sp: | ||
|
|
||
| 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' | ||
|
|
||
| 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 | ||
|
|
||
| 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, { | ||
| assert await async_setup_component(hass, device_tracker.DOMAIN, { | ||
| device_tracker.DOMAIN: { | ||
| CONF_PLATFORM: 'mqtt', | ||
| 'devices': {dev_id: subscription} | ||
| 'devices': {dev_id: topic} | ||
| } | ||
| }) | ||
| fire_mqtt_message(self.hass, topic, location) | ||
| self.hass.block_till_done() | ||
| assert self.hass.states.get(entity_id) is None | ||
| assert mock_sp.call_count == 1 | ||
|
|
||
|
|
||
| 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' | ||
|
|
||
| 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 | ||
|
|
||
|
|
||
| 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' | ||
|
|
||
| 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_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 |
Oops, something went wrong.