Convert device tracker init tests to async (#18640)

This commit is contained in:
Adam Mills 2018-11-23 02:55:25 -05:00 committed by Paulus Schoutsen
parent 98f159a039
commit c99204149c
2 changed files with 457 additions and 416 deletions

View File

@ -0,0 +1,31 @@
"""Collection of helper methods.
All containing methods are legacy helpers that should not be used by new
components. Instead call the service directly.
"""
from homeassistant.components.device_tracker import (
DOMAIN, ATTR_ATTRIBUTES, ATTR_BATTERY, ATTR_GPS, ATTR_GPS_ACCURACY,
ATTR_LOCATION_NAME, ATTR_MAC, ATTR_DEV_ID, ATTR_HOST_NAME, SERVICE_SEE)
from homeassistant.core import callback
from homeassistant.helpers.typing import GPSType, HomeAssistantType
from homeassistant.loader import bind_hass
@callback
@bind_hass
def async_see(hass: HomeAssistantType, mac: str = None, dev_id: str = None,
host_name: str = None, location_name: str = None,
gps: GPSType = None, gps_accuracy=None,
battery: int = None, attributes: dict = None):
"""Call service to notify you see device."""
data = {key: value for key, value in
((ATTR_MAC, mac),
(ATTR_DEV_ID, dev_id),
(ATTR_HOST_NAME, host_name),
(ATTR_LOCATION_NAME, location_name),
(ATTR_GPS, gps),
(ATTR_GPS_ACCURACY, gps_accuracy),
(ATTR_BATTERY, battery)) if value is not None}
if attributes:
data[ATTR_ATTRIBUTES] = attributes
hass.async_add_job(hass.services.async_call(DOMAIN, SERVICE_SEE, data))

View File

@ -3,504 +3,514 @@
import asyncio import asyncio
import json import json
import logging import logging
import unittest from unittest.mock import call
from unittest.mock import call, patch
from datetime import datetime, timedelta from datetime import datetime, timedelta
import os import os
from asynctest import patch
import pytest
from homeassistant.components import zone from homeassistant.components import zone
from homeassistant.core import callback, State from homeassistant.core import callback, State
from homeassistant.setup import setup_component, async_setup_component from homeassistant.setup import async_setup_component
from homeassistant.helpers import discovery from homeassistant.helpers import discovery
from homeassistant.loader import get_component from homeassistant.loader import get_component
from homeassistant.util.async_ import run_coroutine_threadsafe
import homeassistant.util.dt as dt_util import homeassistant.util.dt as dt_util
from homeassistant.const import ( from homeassistant.const import (
ATTR_ENTITY_ID, ATTR_ENTITY_PICTURE, ATTR_FRIENDLY_NAME, ATTR_HIDDEN, ATTR_ENTITY_ID, ATTR_ENTITY_PICTURE, ATTR_FRIENDLY_NAME, ATTR_HIDDEN,
STATE_HOME, STATE_NOT_HOME, CONF_PLATFORM, ATTR_ICON) STATE_HOME, STATE_NOT_HOME, CONF_PLATFORM, ATTR_ICON)
import homeassistant.components.device_tracker as device_tracker import homeassistant.components.device_tracker as device_tracker
from tests.components.device_tracker import common
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.json import JSONEncoder from homeassistant.helpers.json import JSONEncoder
from tests.common import ( from tests.common import (
get_test_home_assistant, fire_time_changed, async_fire_time_changed, patch_yaml_files, assert_setup_component,
patch_yaml_files, assert_setup_component, mock_restore_cache) mock_restore_cache)
import pytest
TEST_PLATFORM = {device_tracker.DOMAIN: {CONF_PLATFORM: 'test'}} TEST_PLATFORM = {device_tracker.DOMAIN: {CONF_PLATFORM: 'test'}}
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
class TestComponentsDeviceTracker(unittest.TestCase): @pytest.fixture
"""Test the Device tracker.""" def yaml_devices(hass):
"""Get a path for storing yaml devices."""
yaml_devices = hass.config.path(device_tracker.YAML_DEVICES)
if os.path.isfile(yaml_devices):
os.remove(yaml_devices)
yield yaml_devices
if os.path.isfile(yaml_devices):
os.remove(yaml_devices)
hass = None # HomeAssistant
yaml_devices = None # type: str
# pylint: disable=invalid-name async def test_is_on(hass):
def setUp(self): """Test is_on method."""
"""Set up things to be run when tests are started.""" entity_id = device_tracker.ENTITY_ID_FORMAT.format('test')
self.hass = get_test_home_assistant()
self.yaml_devices = self.hass.config.path(device_tracker.YAML_DEVICES)
# pylint: disable=invalid-name hass.states.async_set(entity_id, STATE_HOME)
def tearDown(self):
"""Stop everything that was started."""
if os.path.isfile(self.yaml_devices):
os.remove(self.yaml_devices)
self.hass.stop() assert device_tracker.is_on(hass, entity_id)
def test_is_on(self): hass.states.async_set(entity_id, STATE_NOT_HOME)
"""Test is_on method."""
entity_id = device_tracker.ENTITY_ID_FORMAT.format('test')
self.hass.states.set(entity_id, STATE_HOME) assert not device_tracker.is_on(hass, entity_id)
assert device_tracker.is_on(self.hass, entity_id)
self.hass.states.set(entity_id, STATE_NOT_HOME) async def test_reading_broken_yaml_config(hass):
"""Test when known devices contains invalid data."""
files = {'empty.yaml': '',
'nodict.yaml': '100',
'badkey.yaml': '@:\n name: Device',
'noname.yaml': 'my_device:\n',
'allok.yaml': 'My Device:\n name: Device',
'oneok.yaml': ('My Device!:\n name: Device\n'
'bad_device:\n nme: Device')}
args = {'hass': hass, 'consider_home': timedelta(seconds=60)}
with patch_yaml_files(files):
assert await device_tracker.async_load_config(
'empty.yaml', **args) == []
assert await device_tracker.async_load_config(
'nodict.yaml', **args) == []
assert await device_tracker.async_load_config(
'noname.yaml', **args) == []
assert await device_tracker.async_load_config(
'badkey.yaml', **args) == []
assert not device_tracker.is_on(self.hass, entity_id) res = await device_tracker.async_load_config('allok.yaml', **args)
assert len(res) == 1
assert res[0].name == 'Device'
assert res[0].dev_id == 'my_device'
# pylint: disable=no-self-use res = await device_tracker.async_load_config('oneok.yaml', **args)
def test_reading_broken_yaml_config(self): assert len(res) == 1
"""Test when known devices contains invalid data.""" assert res[0].name == 'Device'
files = {'empty.yaml': '', assert res[0].dev_id == 'my_device'
'nodict.yaml': '100',
'badkey.yaml': '@:\n name: Device',
'noname.yaml': 'my_device:\n',
'allok.yaml': 'My Device:\n name: Device',
'oneok.yaml': ('My Device!:\n name: Device\n'
'bad_device:\n nme: Device')}
args = {'hass': self.hass, 'consider_home': timedelta(seconds=60)}
with patch_yaml_files(files):
assert device_tracker.load_config('empty.yaml', **args) == []
assert device_tracker.load_config('nodict.yaml', **args) == []
assert device_tracker.load_config('noname.yaml', **args) == []
assert device_tracker.load_config('badkey.yaml', **args) == []
res = device_tracker.load_config('allok.yaml', **args)
assert len(res) == 1
assert res[0].name == 'Device'
assert res[0].dev_id == 'my_device'
res = device_tracker.load_config('oneok.yaml', **args) async def test_reading_yaml_config(hass, yaml_devices):
assert len(res) == 1 """Test the rendering of the YAML configuration."""
assert res[0].name == 'Device' dev_id = 'test'
assert res[0].dev_id == 'my_device' device = device_tracker.Device(
hass, timedelta(seconds=180), True, dev_id,
'AB:CD:EF:GH:IJ', 'Test name', picture='http://test.picture',
hide_if_away=True, icon='mdi:kettle')
device_tracker.update_config(yaml_devices, dev_id, device)
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(hass, device_tracker.DOMAIN,
TEST_PLATFORM)
config = (await device_tracker.async_load_config(yaml_devices, hass,
device.consider_home))[0]
assert device.dev_id == config.dev_id
assert device.track == config.track
assert device.mac == config.mac
assert device.config_picture == config.config_picture
assert device.away_hide == config.away_hide
assert device.consider_home == config.consider_home
assert device.icon == config.icon
def test_reading_yaml_config(self):
"""Test the rendering of the YAML configuration.""" # pylint: disable=invalid-name
dev_id = 'test' @patch('homeassistant.components.device_tracker._LOGGER.warning')
device = device_tracker.Device( async def test_track_with_duplicate_mac_dev_id(mock_warning, hass):
self.hass, timedelta(seconds=180), True, dev_id, """Test adding duplicate MACs or device IDs to DeviceTracker."""
'AB:CD:EF:GH:IJ', 'Test name', picture='http://test.picture', devices = [
hide_if_away=True, icon='mdi:kettle') device_tracker.Device(hass, True, True, 'my_device', 'AB:01',
device_tracker.update_config(self.yaml_devices, dev_id, device) 'My device', None, None, False),
device_tracker.Device(hass, True, True, 'your_device',
'AB:01', 'Your device', None, None, False)]
device_tracker.DeviceTracker(hass, False, True, {}, devices)
_LOGGER.debug(mock_warning.call_args_list)
assert mock_warning.call_count == 1, \
"The only warning call should be duplicates (check DEBUG)"
args, _ = mock_warning.call_args
assert 'Duplicate device MAC' in args[0], \
'Duplicate MAC warning expected'
mock_warning.reset_mock()
devices = [
device_tracker.Device(hass, True, True, 'my_device',
'AB:01', 'My device', None, None, False),
device_tracker.Device(hass, True, True, 'my_device',
None, 'Your device', None, None, False)]
device_tracker.DeviceTracker(hass, False, True, {}, devices)
_LOGGER.debug(mock_warning.call_args_list)
assert mock_warning.call_count == 1, \
"The only warning call should be duplicates (check DEBUG)"
args, _ = mock_warning.call_args
assert 'Duplicate device IDs' in args[0], \
'Duplicate device IDs warning expected'
async def test_setup_without_yaml_file(hass):
"""Test with no YAML file."""
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(hass, device_tracker.DOMAIN,
TEST_PLATFORM)
async def test_gravatar(hass):
"""Test the Gravatar generation."""
dev_id = 'test'
device = device_tracker.Device(
hass, timedelta(seconds=180), True, dev_id,
'AB:CD:EF:GH:IJ', 'Test name', gravatar='test@example.com')
gravatar_url = ("https://www.gravatar.com/avatar/"
"55502f40dc8b7c769880b10874abc9d0.jpg?s=80&d=wavatar")
assert device.config_picture == gravatar_url
async def test_gravatar_and_picture(hass):
"""Test that Gravatar overrides picture."""
dev_id = 'test'
device = device_tracker.Device(
hass, timedelta(seconds=180), True, dev_id,
'AB:CD:EF:GH:IJ', 'Test name', picture='http://test.picture',
gravatar='test@example.com')
gravatar_url = ("https://www.gravatar.com/avatar/"
"55502f40dc8b7c769880b10874abc9d0.jpg?s=80&d=wavatar")
assert device.config_picture == gravatar_url
@patch(
'homeassistant.components.device_tracker.DeviceTracker.see')
@patch(
'homeassistant.components.device_tracker.demo.setup_scanner',
autospec=True)
async def test_discover_platform(mock_demo_setup_scanner, mock_see, hass):
"""Test discovery of device_tracker demo platform."""
assert device_tracker.DOMAIN not in hass.config.components
await discovery.async_load_platform(
hass, device_tracker.DOMAIN, 'demo', {'test_key': 'test_val'},
{'demo': {}})
await hass.async_block_till_done()
assert device_tracker.DOMAIN in hass.config.components
assert mock_demo_setup_scanner.called
assert mock_demo_setup_scanner.call_args[0] == (
hass, {}, mock_see, {'test_key': 'test_val'})
async def test_update_stale(hass):
"""Test stalled update."""
scanner = get_component(hass, 'device_tracker.test').SCANNER
scanner.reset()
scanner.come_home('DEV1')
register_time = datetime(2015, 9, 15, 23, tzinfo=dt_util.UTC)
scan_time = datetime(2015, 9, 15, 23, 1, tzinfo=dt_util.UTC)
with patch('homeassistant.components.device_tracker.dt_util.utcnow',
return_value=register_time):
with assert_setup_component(1, device_tracker.DOMAIN): with assert_setup_component(1, device_tracker.DOMAIN):
assert setup_component(self.hass, device_tracker.DOMAIN, assert await async_setup_component(hass, device_tracker.DOMAIN, {
TEST_PLATFORM) device_tracker.DOMAIN: {
config = device_tracker.load_config(self.yaml_devices, self.hass, CONF_PLATFORM: 'test',
device.consider_home)[0] device_tracker.CONF_CONSIDER_HOME: 59,
assert device.dev_id == config.dev_id }})
assert device.track == config.track await hass.async_block_till_done()
assert device.mac == config.mac
assert device.config_picture == config.config_picture
assert device.away_hide == config.away_hide
assert device.consider_home == config.consider_home
assert device.icon == config.icon
# pylint: disable=invalid-name assert STATE_HOME == \
@patch('homeassistant.components.device_tracker._LOGGER.warning') hass.states.get('device_tracker.dev1').state
def test_track_with_duplicate_mac_dev_id(self, mock_warning):
"""Test adding duplicate MACs or device IDs to DeviceTracker."""
devices = [
device_tracker.Device(self.hass, True, True, 'my_device', 'AB:01',
'My device', None, None, False),
device_tracker.Device(self.hass, True, True, 'your_device',
'AB:01', 'Your device', None, None, False)]
device_tracker.DeviceTracker(self.hass, False, True, {}, devices)
_LOGGER.debug(mock_warning.call_args_list)
assert mock_warning.call_count == 1, \
"The only warning call should be duplicates (check DEBUG)"
args, _ = mock_warning.call_args
assert 'Duplicate device MAC' in args[0], \
'Duplicate MAC warning expected'
mock_warning.reset_mock() scanner.leave_home('DEV1')
devices = [
device_tracker.Device(self.hass, True, True, 'my_device',
'AB:01', 'My device', None, None, False),
device_tracker.Device(self.hass, True, True, 'my_device',
None, 'Your device', None, None, False)]
device_tracker.DeviceTracker(self.hass, False, True, {}, devices)
_LOGGER.debug(mock_warning.call_args_list) with patch('homeassistant.components.device_tracker.dt_util.utcnow',
assert mock_warning.call_count == 1, \ return_value=scan_time):
"The only warning call should be duplicates (check DEBUG)" async_fire_time_changed(hass, scan_time)
args, _ = mock_warning.call_args await hass.async_block_till_done()
assert 'Duplicate device IDs' in args[0], \
'Duplicate device IDs warning expected'
def test_setup_without_yaml_file(self): assert STATE_NOT_HOME == \
"""Test with no YAML file.""" hass.states.get('device_tracker.dev1').state
with assert_setup_component(1, device_tracker.DOMAIN):
assert setup_component(self.hass, device_tracker.DOMAIN,
TEST_PLATFORM)
def test_gravatar(self):
"""Test the Gravatar generation."""
dev_id = 'test'
device = device_tracker.Device(
self.hass, timedelta(seconds=180), True, dev_id,
'AB:CD:EF:GH:IJ', 'Test name', gravatar='test@example.com')
gravatar_url = ("https://www.gravatar.com/avatar/"
"55502f40dc8b7c769880b10874abc9d0.jpg?s=80&d=wavatar")
assert device.config_picture == gravatar_url
def test_gravatar_and_picture(self): async def test_entity_attributes(hass, yaml_devices):
"""Test that Gravatar overrides picture.""" """Test the entity attributes."""
dev_id = 'test' dev_id = 'test_entity'
device = device_tracker.Device( entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
self.hass, timedelta(seconds=180), True, dev_id, friendly_name = 'Paulus'
'AB:CD:EF:GH:IJ', 'Test name', picture='http://test.picture', picture = 'http://placehold.it/200x200'
gravatar='test@example.com') icon = 'mdi:kettle'
gravatar_url = ("https://www.gravatar.com/avatar/"
"55502f40dc8b7c769880b10874abc9d0.jpg?s=80&d=wavatar")
assert device.config_picture == gravatar_url
@patch( device = device_tracker.Device(
'homeassistant.components.device_tracker.DeviceTracker.see') hass, timedelta(seconds=180), True, dev_id, None,
@patch( friendly_name, picture, hide_if_away=True, icon=icon)
'homeassistant.components.device_tracker.demo.setup_scanner', device_tracker.update_config(yaml_devices, dev_id, device)
autospec=True)
def test_discover_platform(self, mock_demo_setup_scanner, mock_see):
"""Test discovery of device_tracker demo platform."""
assert device_tracker.DOMAIN not in self.hass.config.components
discovery.load_platform(
self.hass, device_tracker.DOMAIN, 'demo', {'test_key': 'test_val'},
{'demo': {}})
self.hass.block_till_done()
assert device_tracker.DOMAIN in self.hass.config.components
assert mock_demo_setup_scanner.called
assert mock_demo_setup_scanner.call_args[0] == (
self.hass, {}, mock_see, {'test_key': 'test_val'})
def test_update_stale(self): with assert_setup_component(1, device_tracker.DOMAIN):
"""Test stalled update.""" assert await async_setup_component(hass, device_tracker.DOMAIN,
scanner = get_component(self.hass, 'device_tracker.test').SCANNER TEST_PLATFORM)
scanner.reset()
scanner.come_home('DEV1')
register_time = datetime(2015, 9, 15, 23, tzinfo=dt_util.UTC) attrs = hass.states.get(entity_id).attributes
scan_time = datetime(2015, 9, 15, 23, 1, tzinfo=dt_util.UTC)
with patch('homeassistant.components.device_tracker.dt_util.utcnow', assert friendly_name == attrs.get(ATTR_FRIENDLY_NAME)
return_value=register_time): assert icon == attrs.get(ATTR_ICON)
with assert_setup_component(1, device_tracker.DOMAIN): assert picture == attrs.get(ATTR_ENTITY_PICTURE)
assert setup_component(self.hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'test',
device_tracker.CONF_CONSIDER_HOME: 59,
}})
self.hass.block_till_done()
assert STATE_HOME == \
self.hass.states.get('device_tracker.dev1').state
scanner.leave_home('DEV1') async def test_device_hidden(hass, yaml_devices):
"""Test hidden devices."""
dev_id = 'test_entity'
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
device = device_tracker.Device(
hass, timedelta(seconds=180), True, dev_id, None,
hide_if_away=True)
device_tracker.update_config(yaml_devices, dev_id, device)
with patch('homeassistant.components.device_tracker.dt_util.utcnow', scanner = get_component(hass, 'device_tracker.test').SCANNER
return_value=scan_time): scanner.reset()
fire_time_changed(self.hass, scan_time)
self.hass.block_till_done()
assert STATE_NOT_HOME == \ with assert_setup_component(1, device_tracker.DOMAIN):
self.hass.states.get('device_tracker.dev1').state assert await async_setup_component(hass, device_tracker.DOMAIN,
TEST_PLATFORM)
def test_entity_attributes(self): assert hass.states.get(entity_id).attributes.get(ATTR_HIDDEN)
"""Test the entity attributes."""
dev_id = 'test_entity'
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
friendly_name = 'Paulus'
picture = 'http://placehold.it/200x200'
icon = 'mdi:kettle'
device = device_tracker.Device(
self.hass, timedelta(seconds=180), True, dev_id, None,
friendly_name, picture, hide_if_away=True, icon=icon)
device_tracker.update_config(self.yaml_devices, dev_id, device)
with assert_setup_component(1, device_tracker.DOMAIN): async def test_group_all_devices(hass, yaml_devices):
assert setup_component(self.hass, device_tracker.DOMAIN, """Test grouping of devices."""
TEST_PLATFORM) dev_id = 'test_entity'
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id)
device = device_tracker.Device(
hass, timedelta(seconds=180), True, dev_id, None,
hide_if_away=True)
device_tracker.update_config(yaml_devices, dev_id, device)
attrs = self.hass.states.get(entity_id).attributes scanner = get_component(hass, 'device_tracker.test').SCANNER
scanner.reset()
assert friendly_name == attrs.get(ATTR_FRIENDLY_NAME) with assert_setup_component(1, device_tracker.DOMAIN):
assert icon == attrs.get(ATTR_ICON) assert await async_setup_component(hass, device_tracker.DOMAIN,
assert picture == attrs.get(ATTR_ENTITY_PICTURE) TEST_PLATFORM)
await hass.async_block_till_done()
def test_device_hidden(self): state = hass.states.get(device_tracker.ENTITY_ID_ALL_DEVICES)
"""Test hidden devices.""" assert state is not None
dev_id = 'test_entity' assert STATE_NOT_HOME == state.state
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id) assert (entity_id,) == state.attributes.get(ATTR_ENTITY_ID)
device = device_tracker.Device(
self.hass, timedelta(seconds=180), True, dev_id, None,
hide_if_away=True)
device_tracker.update_config(self.yaml_devices, dev_id, device)
scanner = get_component(self.hass, 'device_tracker.test').SCANNER
scanner.reset()
with assert_setup_component(1, device_tracker.DOMAIN): @patch('homeassistant.components.device_tracker.DeviceTracker.async_see')
assert setup_component(self.hass, device_tracker.DOMAIN, async def test_see_service(mock_see, hass):
TEST_PLATFORM) """Test the see service with a unicode dev_id and NO MAC."""
with assert_setup_component(1, device_tracker.DOMAIN):
assert self.hass.states.get(entity_id) \ assert await async_setup_component(hass, device_tracker.DOMAIN,
.attributes.get(ATTR_HIDDEN) TEST_PLATFORM)
params = {
def test_group_all_devices(self): 'dev_id': 'some_device',
"""Test grouping of devices.""" 'host_name': 'example.com',
dev_id = 'test_entity' 'location_name': 'Work',
entity_id = device_tracker.ENTITY_ID_FORMAT.format(dev_id) 'gps': [.3, .8],
device = device_tracker.Device( 'attributes': {
self.hass, timedelta(seconds=180), True, dev_id, None, 'test': 'test'
hide_if_away=True)
device_tracker.update_config(self.yaml_devices, dev_id, device)
scanner = get_component(self.hass, 'device_tracker.test').SCANNER
scanner.reset()
with assert_setup_component(1, device_tracker.DOMAIN):
assert setup_component(self.hass, device_tracker.DOMAIN,
TEST_PLATFORM)
self.hass.block_till_done()
state = self.hass.states.get(device_tracker.ENTITY_ID_ALL_DEVICES)
assert state is not None
assert STATE_NOT_HOME == state.state
assert (entity_id,) == state.attributes.get(ATTR_ENTITY_ID)
@patch('homeassistant.components.device_tracker.DeviceTracker.async_see')
def test_see_service(self, mock_see):
"""Test the see service with a unicode dev_id and NO MAC."""
with assert_setup_component(1, device_tracker.DOMAIN):
assert setup_component(self.hass, device_tracker.DOMAIN,
TEST_PLATFORM)
params = {
'dev_id': 'some_device',
'host_name': 'example.com',
'location_name': 'Work',
'gps': [.3, .8],
'attributes': {
'test': 'test'
}
} }
device_tracker.see(self.hass, **params) }
self.hass.block_till_done() common.async_see(hass, **params)
assert mock_see.call_count == 1 await hass.async_block_till_done()
assert mock_see.call_count == 1 assert mock_see.call_count == 1
assert mock_see.call_args == call(**params) assert mock_see.call_count == 1
assert mock_see.call_args == call(**params)
mock_see.reset_mock() mock_see.reset_mock()
params['dev_id'] += chr(233) # e' acute accent from icloud params['dev_id'] += chr(233) # e' acute accent from icloud
device_tracker.see(self.hass, **params) common.async_see(hass, **params)
self.hass.block_till_done() await hass.async_block_till_done()
assert mock_see.call_count == 1 assert mock_see.call_count == 1
assert mock_see.call_count == 1 assert mock_see.call_count == 1
assert mock_see.call_args == call(**params) assert mock_see.call_args == call(**params)
def test_new_device_event_fired(self):
"""Test that the device tracker will fire an event."""
with assert_setup_component(1, device_tracker.DOMAIN):
assert setup_component(self.hass, device_tracker.DOMAIN,
TEST_PLATFORM)
test_events = []
@callback async def test_new_device_event_fired(hass):
def listener(event): """Test that the device tracker will fire an event."""
"""Record that our event got called.""" with assert_setup_component(1, device_tracker.DOMAIN):
test_events.append(event) assert await async_setup_component(hass, device_tracker.DOMAIN,
TEST_PLATFORM)
test_events = []
self.hass.bus.listen("device_tracker_new_device", listener) @callback
def listener(event):
"""Record that our event got called."""
test_events.append(event)
device_tracker.see(self.hass, 'mac_1', host_name='hello') hass.bus.async_listen("device_tracker_new_device", listener)
device_tracker.see(self.hass, 'mac_1', host_name='hello')
self.hass.block_till_done() common.async_see(hass, 'mac_1', host_name='hello')
common.async_see(hass, 'mac_1', host_name='hello')
assert len(test_events) == 1 await hass.async_block_till_done()
# Assert we can serialize the event assert len(test_events) == 1
json.dumps(test_events[0].as_dict(), cls=JSONEncoder)
assert test_events[0].data == { # Assert we can serialize the event
'entity_id': 'device_tracker.hello', json.dumps(test_events[0].as_dict(), cls=JSONEncoder)
'host_name': 'hello',
'mac': 'MAC_1', assert test_events[0].data == {
'entity_id': 'device_tracker.hello',
'host_name': 'hello',
'mac': 'MAC_1',
}
# pylint: disable=invalid-name
async def test_not_write_duplicate_yaml_keys(hass, yaml_devices):
"""Test that the device tracker will not generate invalid YAML."""
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(hass, device_tracker.DOMAIN,
TEST_PLATFORM)
common.async_see(hass, 'mac_1', host_name='hello')
common.async_see(hass, 'mac_2', host_name='hello')
await hass.async_block_till_done()
config = await device_tracker.async_load_config(yaml_devices, hass,
timedelta(seconds=0))
assert len(config) == 2
# pylint: disable=invalid-name
async def test_not_allow_invalid_dev_id(hass, yaml_devices):
"""Test that the device tracker will not allow invalid dev ids."""
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(hass, device_tracker.DOMAIN,
TEST_PLATFORM)
common.async_see(hass, dev_id='hello-world')
config = await device_tracker.async_load_config(yaml_devices, hass,
timedelta(seconds=0))
assert len(config) == 0
async def test_see_state(hass, yaml_devices):
"""Test device tracker see records state correctly."""
assert await async_setup_component(hass, device_tracker.DOMAIN,
TEST_PLATFORM)
params = {
'mac': 'AA:BB:CC:DD:EE:FF',
'dev_id': 'some_device',
'host_name': 'example.com',
'location_name': 'Work',
'gps': [.3, .8],
'gps_accuracy': 1,
'battery': 100,
'attributes': {
'test': 'test',
'number': 1,
},
}
common.async_see(hass, **params)
await hass.async_block_till_done()
config = await device_tracker.async_load_config(yaml_devices, hass,
timedelta(seconds=0))
assert len(config) == 1
state = hass.states.get('device_tracker.examplecom')
attrs = state.attributes
assert state.state == 'Work'
assert state.object_id == 'examplecom'
assert state.name == 'example.com'
assert attrs['friendly_name'] == 'example.com'
assert attrs['battery'] == 100
assert attrs['latitude'] == 0.3
assert attrs['longitude'] == 0.8
assert attrs['test'] == 'test'
assert attrs['gps_accuracy'] == 1
assert attrs['source_type'] == 'gps'
assert attrs['number'] == 1
async def test_see_passive_zone_state(hass):
"""Test that the device tracker sets gps for passive trackers."""
register_time = datetime(2015, 9, 15, 23, tzinfo=dt_util.UTC)
scan_time = datetime(2015, 9, 15, 23, 1, tzinfo=dt_util.UTC)
with assert_setup_component(1, zone.DOMAIN):
zone_info = {
'name': 'Home',
'latitude': 1,
'longitude': 2,
'radius': 250,
'passive': False
} }
# pylint: disable=invalid-name await async_setup_component(hass, zone.DOMAIN, {
def test_not_write_duplicate_yaml_keys(self): 'zone': zone_info
"""Test that the device tracker will not generate invalid YAML.""" })
scanner = get_component(hass, 'device_tracker.test').SCANNER
scanner.reset()
scanner.come_home('dev1')
with patch('homeassistant.components.device_tracker.dt_util.utcnow',
return_value=register_time):
with assert_setup_component(1, device_tracker.DOMAIN): with assert_setup_component(1, device_tracker.DOMAIN):
assert setup_component(self.hass, device_tracker.DOMAIN, assert await async_setup_component(hass, device_tracker.DOMAIN, {
TEST_PLATFORM) device_tracker.DOMAIN: {
CONF_PLATFORM: 'test',
device_tracker.CONF_CONSIDER_HOME: 59,
}})
await hass.async_block_till_done()
device_tracker.see(self.hass, 'mac_1', host_name='hello') state = hass.states.get('device_tracker.dev1')
device_tracker.see(self.hass, 'mac_2', host_name='hello') attrs = state.attributes
assert STATE_HOME == state.state
assert state.object_id == 'dev1'
assert state.name == 'dev1'
assert attrs.get('friendly_name') == 'dev1'
assert attrs.get('latitude') == 1
assert attrs.get('longitude') == 2
assert attrs.get('gps_accuracy') == 0
assert attrs.get('source_type') == \
device_tracker.SOURCE_TYPE_ROUTER
self.hass.block_till_done() scanner.leave_home('dev1')
config = device_tracker.load_config(self.yaml_devices, self.hass, with patch('homeassistant.components.device_tracker.dt_util.utcnow',
timedelta(seconds=0)) return_value=scan_time):
assert len(config) == 2 async_fire_time_changed(hass, scan_time)
await hass.async_block_till_done()
# pylint: disable=invalid-name state = hass.states.get('device_tracker.dev1')
def test_not_allow_invalid_dev_id(self): attrs = state.attributes
"""Test that the device tracker will not allow invalid dev ids.""" assert STATE_NOT_HOME == state.state
with assert_setup_component(1, device_tracker.DOMAIN): assert state.object_id == 'dev1'
assert setup_component(self.hass, device_tracker.DOMAIN, assert state.name == 'dev1'
TEST_PLATFORM) assert attrs.get('friendly_name') == 'dev1'
assert attrs.get('latitude')is None
assert attrs.get('longitude')is None
assert attrs.get('gps_accuracy')is None
assert attrs.get('source_type') == \
device_tracker.SOURCE_TYPE_ROUTER
device_tracker.see(self.hass, dev_id='hello-world')
config = device_tracker.load_config(self.yaml_devices, self.hass, @patch('homeassistant.components.device_tracker._LOGGER.warning')
timedelta(seconds=0)) async def test_see_failures(mock_warning, hass, yaml_devices):
assert len(config) == 0 """Test that the device tracker see failures."""
tracker = device_tracker.DeviceTracker(
hass, timedelta(seconds=60), 0, {}, [])
def test_see_state(self): # MAC is not a string (but added)
"""Test device tracker see records state correctly.""" await tracker.async_see(mac=567, host_name="Number MAC")
assert setup_component(self.hass, device_tracker.DOMAIN,
TEST_PLATFORM)
params = { # No device id or MAC(not added)
'mac': 'AA:BB:CC:DD:EE:FF', with pytest.raises(HomeAssistantError):
'dev_id': 'some_device', await tracker.async_see()
'host_name': 'example.com', assert mock_warning.call_count == 0
'location_name': 'Work',
'gps': [.3, .8],
'gps_accuracy': 1,
'battery': 100,
'attributes': {
'test': 'test',
'number': 1,
},
}
device_tracker.see(self.hass, **params) # Ignore gps on invalid GPS (both added & warnings)
self.hass.block_till_done() await tracker.async_see(mac='mac_1_bad_gps', gps=1)
await tracker.async_see(mac='mac_2_bad_gps', gps=[1])
await tracker.async_see(mac='mac_3_bad_gps', gps='gps')
await hass.async_block_till_done()
config = await device_tracker.async_load_config(yaml_devices, hass,
timedelta(seconds=0))
assert mock_warning.call_count == 3
config = device_tracker.load_config(self.yaml_devices, self.hass, assert len(config) == 4
timedelta(seconds=0))
assert len(config) == 1
state = self.hass.states.get('device_tracker.examplecom')
attrs = state.attributes
assert state.state == 'Work'
assert state.object_id == 'examplecom'
assert state.name == 'example.com'
assert attrs['friendly_name'] == 'example.com'
assert attrs['battery'] == 100
assert attrs['latitude'] == 0.3
assert attrs['longitude'] == 0.8
assert attrs['test'] == 'test'
assert attrs['gps_accuracy'] == 1
assert attrs['source_type'] == 'gps'
assert attrs['number'] == 1
def test_see_passive_zone_state(self):
"""Test that the device tracker sets gps for passive trackers."""
register_time = datetime(2015, 9, 15, 23, tzinfo=dt_util.UTC)
scan_time = datetime(2015, 9, 15, 23, 1, tzinfo=dt_util.UTC)
with assert_setup_component(1, zone.DOMAIN):
zone_info = {
'name': 'Home',
'latitude': 1,
'longitude': 2,
'radius': 250,
'passive': False
}
setup_component(self.hass, zone.DOMAIN, {
'zone': zone_info
})
scanner = get_component(self.hass, 'device_tracker.test').SCANNER
scanner.reset()
scanner.come_home('dev1')
with patch('homeassistant.components.device_tracker.dt_util.utcnow',
return_value=register_time):
with assert_setup_component(1, device_tracker.DOMAIN):
assert setup_component(self.hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'test',
device_tracker.CONF_CONSIDER_HOME: 59,
}})
self.hass.block_till_done()
state = self.hass.states.get('device_tracker.dev1')
attrs = state.attributes
assert STATE_HOME == state.state
assert state.object_id == 'dev1'
assert state.name == 'dev1'
assert attrs.get('friendly_name') == 'dev1'
assert attrs.get('latitude') == 1
assert attrs.get('longitude') == 2
assert attrs.get('gps_accuracy') == 0
assert attrs.get('source_type') == \
device_tracker.SOURCE_TYPE_ROUTER
scanner.leave_home('dev1')
with patch('homeassistant.components.device_tracker.dt_util.utcnow',
return_value=scan_time):
fire_time_changed(self.hass, scan_time)
self.hass.block_till_done()
state = self.hass.states.get('device_tracker.dev1')
attrs = state.attributes
assert STATE_NOT_HOME == state.state
assert state.object_id == 'dev1'
assert state.name == 'dev1'
assert attrs.get('friendly_name') == 'dev1'
assert attrs.get('latitude')is None
assert attrs.get('longitude')is None
assert attrs.get('gps_accuracy')is None
assert attrs.get('source_type') == \
device_tracker.SOURCE_TYPE_ROUTER
@patch('homeassistant.components.device_tracker._LOGGER.warning')
def test_see_failures(self, mock_warning):
"""Test that the device tracker see failures."""
tracker = device_tracker.DeviceTracker(
self.hass, timedelta(seconds=60), 0, {}, [])
# MAC is not a string (but added)
tracker.see(mac=567, host_name="Number MAC")
# No device id or MAC(not added)
with pytest.raises(HomeAssistantError):
run_coroutine_threadsafe(
tracker.async_see(), self.hass.loop).result()
assert mock_warning.call_count == 0
# Ignore gps on invalid GPS (both added & warnings)
tracker.see(mac='mac_1_bad_gps', gps=1)
tracker.see(mac='mac_2_bad_gps', gps=[1])
tracker.see(mac='mac_3_bad_gps', gps='gps')
self.hass.block_till_done()
config = device_tracker.load_config(self.yaml_devices, self.hass,
timedelta(seconds=0))
assert mock_warning.call_count == 3
assert len(config) == 4
@asyncio.coroutine @asyncio.coroutine