Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fixup device_tracekt.mqtt voluptuous & unit tests ()

pull/3945/head
Johann Kellerman 2016-10-19 03:10:28 +02:00 committed by Paulus Schoutsen
parent 7da47852d4
commit 57f32fa629
6 changed files with 137 additions and 105 deletions
homeassistant
components
helpers
tests/components/device_tracker

16
homeassistant/components/device_tracker/__init__.py
View File

@ -68,15 +68,11 @@ ATTR_ATTRIBUTES = 'attributes'
PLATFORM_SCHEMA = cv.PLATFORM_SCHEMA.extend({
vol.Optional(CONF_SCAN_INTERVAL): cv.positive_int, # seconds
}, extra=vol.ALLOW_EXTRA)
_CONFIG_SCHEMA = vol.Schema({DOMAIN: vol.All(cv.ensure_list, [
vol.Schema({
vol.Optional(CONF_TRACK_NEW, default=DEFAULT_TRACK_NEW): cv.boolean,
vol.Optional(
CONF_CONSIDER_HOME, default=timedelta(seconds=180)): vol.All(
cv.time_period, cv.positive_timedelta)
}, extra=vol.ALLOW_EXTRA)])}, extra=vol.ALLOW_EXTRA)
vol.Optional(CONF_TRACK_NEW, default=DEFAULT_TRACK_NEW): cv.boolean,
vol.Optional(CONF_CONSIDER_HOME,
default=timedelta(seconds=DEFAULT_CONSIDER_HOME)): vol.All(
cv.time_period, cv.positive_timedelta)
})
DISCOVERY_PLATFORMS = {
SERVICE_NETGEAR: 'netgear',
@ -116,7 +112,7 @@ def setup(hass: HomeAssistantType, config: ConfigType):
yaml_path = hass.config.path(YAML_DEVICES)
try:
conf = _CONFIG_SCHEMA(config).get(DOMAIN, [])
conf = config.get(DOMAIN, [])
except vol.Invalid as ex:
log_exception(ex, DOMAIN, config)
return False

3
homeassistant/components/device_tracker/mqtt.py
View File

@ -11,13 +11,14 @@ import voluptuous as vol
import homeassistant.components.mqtt as mqtt
from homeassistant.const import CONF_DEVICES
from homeassistant.components.mqtt import CONF_QOS
from homeassistant.components.device_tracker import PLATFORM_SCHEMA
import homeassistant.helpers.config_validation as cv
DEPENDENCIES = ['mqtt']
_LOGGER = logging.getLogger(__name__)
PLATFORM_SCHEMA = mqtt.MQTT_BASE_PLATFORM_SCHEMA.extend({
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(mqtt.SCHEMA_BASE).extend({
vol.Required(CONF_DEVICES): {cv.string: mqtt.valid_subscribe_topic},
})

12
homeassistant/components/mqtt/__init__.py
View File

@ -18,8 +18,7 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import template, config_validation as cv
from homeassistant.helpers.event import threaded_listener_factory
from homeassistant.const import (
EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP,
CONF_PLATFORM, CONF_SCAN_INTERVAL, CONF_VALUE_TEMPLATE)
EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP, CONF_VALUE_TEMPLATE)
_LOGGER = logging.getLogger(__name__)
@ -107,12 +106,11 @@ CONFIG_SCHEMA = vol.Schema({
}),
}, extra=vol.ALLOW_EXTRA)
MQTT_BASE_PLATFORM_SCHEMA = vol.Schema({
vol.Required(CONF_PLATFORM): DOMAIN,
vol.Optional(CONF_SCAN_INTERVAL):
vol.All(vol.Coerce(int), vol.Range(min=1)),
SCHEMA_BASE = {
vol.Optional(CONF_QOS, default=DEFAULT_QOS): _VALID_QOS_SCHEMA,
})
}
MQTT_BASE_PLATFORM_SCHEMA = cv.PLATFORM_SCHEMA.extend(SCHEMA_BASE)
# Sensor type platforms subscribe to MQTT events
MQTT_RO_PLATFORM_SCHEMA = MQTT_BASE_PLATFORM_SCHEMA.extend({

3
homeassistant/helpers/config_validation.py
View File

@ -358,7 +358,8 @@ def key_dependency(key, dependency):
PLATFORM_SCHEMA = vol.Schema({
vol.Required(CONF_PLATFORM): string,
CONF_SCAN_INTERVAL: vol.All(vol.Coerce(int), vol.Range(min=1)),
vol.Optional(CONF_SCAN_INTERVAL):
vol.All(vol.Coerce(int), vol.Range(min=1)),
}, extra=vol.ALLOW_EXTRA)
EVENT_SCHEMA = vol.Schema({

14
tests/components/device_tracker/test_asuswrt.py
View File

@ -1,5 +1,6 @@
"""The tests for the ASUSWRT device tracker platform."""
import os
from datetime import timedelta
import unittest
from unittest import mock
@ -7,8 +8,11 @@ import voluptuous as vol
from homeassistant.bootstrap import setup_component
from homeassistant.components import device_tracker
from homeassistant.components.device_tracker import (
CONF_CONSIDER_HOME, CONF_TRACK_NEW)
from homeassistant.components.device_tracker.asuswrt import (
CONF_PROTOCOL, CONF_MODE, CONF_PUB_KEY, PLATFORM_SCHEMA, DOMAIN)
CONF_PROTOCOL, CONF_MODE, CONF_PUB_KEY, DOMAIN,
PLATFORM_SCHEMA)
from homeassistant.const import (CONF_PLATFORM, CONF_PASSWORD, CONF_USERNAME,
CONF_HOST)
@ -70,7 +74,9 @@ class TestComponentsDeviceTrackerASUSWRT(unittest.TestCase):
CONF_PLATFORM: 'asuswrt',
CONF_HOST: 'fake_host',
CONF_USERNAME: 'fake_user',
CONF_PASSWORD: 'fake_pass'
CONF_PASSWORD: 'fake_pass',
CONF_TRACK_NEW: True,
CONF_CONSIDER_HOME: timedelta(seconds=180)
}
}
@ -93,7 +99,9 @@ class TestComponentsDeviceTrackerASUSWRT(unittest.TestCase):
CONF_PLATFORM: 'asuswrt',
CONF_HOST: 'fake_host',
CONF_USERNAME: 'fake_user',
CONF_PUB_KEY: FAKEFILE
CONF_PUB_KEY: FAKEFILE,
CONF_TRACK_NEW: True,
CONF_CONSIDER_HOME: timedelta(seconds=180)
}
}

194
tests/components/device_tracker/test_owntracks.py
View File

@ -12,7 +12,8 @@ from homeassistant.const import (STATE_NOT_HOME, CONF_PLATFORM)
import homeassistant.components.device_tracker.owntracks as owntracks
from tests.common import (
get_test_home_assistant, mock_mqtt_component, fire_mqtt_message)
assert_setup_component, get_test_home_assistant, mock_mqtt_component,
fire_mqtt_message)
USER = 'greg'
DEVICE = 'phone'
@ -207,20 +208,60 @@ MOCK_ENCRYPTED_LOCATION_MESSAGE = {
}
class TestDeviceTrackerOwnTracks(unittest.TestCase):
class BaseMQTT(unittest.TestCase):
"""Base MQTT assert functions."""
hass = None
def send_message(self, topic, message, corrupt=False):
"""Test the sending of a message."""
str_message = json.dumps(message)
if corrupt:
mod_message = BAD_JSON_PREFIX + str_message + BAD_JSON_SUFFIX
else:
mod_message = str_message
fire_mqtt_message(self.hass, topic, mod_message)
self.hass.block_till_done()
def assert_location_state(self, location):
"""Test the assertion of a location state."""
state = self.hass.states.get(DEVICE_TRACKER_STATE)
self.assertEqual(state.state, location)
def assert_location_latitude(self, latitude):
"""Test the assertion of a location latitude."""
state = self.hass.states.get(DEVICE_TRACKER_STATE)
self.assertEqual(state.attributes.get('latitude'), latitude)
def assert_location_longitude(self, longitude):
"""Test the assertion of a location longitude."""
state = self.hass.states.get(DEVICE_TRACKER_STATE)
self.assertEqual(state.attributes.get('longitude'), longitude)
def assert_location_accuracy(self, accuracy):
"""Test the assertion of a location accuracy."""
state = self.hass.states.get(DEVICE_TRACKER_STATE)
self.assertEqual(state.attributes.get('gps_accuracy'), accuracy)
# pylint: disable=too-many-public-methods
class TestDeviceTrackerOwnTracks(BaseMQTT):
"""Test the OwnTrack sensor."""
def setup_method(self, method):
# pylint: disable=invalid-name
def setup_method(self, _):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
mock_mqtt_component(self.hass)
self.assertTrue(setup_component(self.hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_MAX_GPS_ACCURACY: 200,
CONF_WAYPOINT_IMPORT: True,
CONF_WAYPOINT_WHITELIST: ['jon', 'greg']
}}))
with assert_setup_component(1, device_tracker.DOMAIN):
assert setup_component(self.hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_MAX_GPS_ACCURACY: 200,
CONF_WAYPOINT_IMPORT: True,
CONF_WAYPOINT_WHITELIST: ['jon', 'greg']
}})
self.hass.states.set(
'zone.inner', 'zoning',
@ -254,7 +295,7 @@ class TestDeviceTrackerOwnTracks(unittest.TestCase):
owntracks.REGIONS_ENTERED = defaultdict(list)
owntracks.MOBILE_BEACONS_ACTIVE = defaultdict(list)
def teardown_method(self, method):
def teardown_method(self, _):
"""Stop everything that was started."""
self.hass.stop()
@ -263,40 +304,6 @@ class TestDeviceTrackerOwnTracks(unittest.TestCase):
except FileNotFoundError:
pass
def mock_see(**kwargs):
"""Fake see method for owntracks."""
return
def send_message(self, topic, message, corrupt=False):
"""Test the sending of a message."""
str_message = json.dumps(message)
if corrupt:
mod_message = BAD_JSON_PREFIX + str_message + BAD_JSON_SUFFIX
else:
mod_message = str_message
fire_mqtt_message(self.hass, topic, mod_message)
self.hass.block_till_done()
def assert_location_state(self, location):
"""Test the assertion of a location state."""
state = self.hass.states.get(DEVICE_TRACKER_STATE)
self.assertEqual(state.state, location)
def assert_location_latitude(self, latitude):
"""Test the assertion of a location latitude."""
state = self.hass.states.get(DEVICE_TRACKER_STATE)
self.assertEqual(state.attributes.get('latitude'), latitude)
def assert_location_longitude(self, longitude):
"""Test the assertion of a location longitude."""
state = self.hass.states.get(DEVICE_TRACKER_STATE)
self.assertEqual(state.attributes.get('longitude'), longitude)
def assert_location_accuracy(self, accuracy):
"""Test the assertion of a location accuracy."""
state = self.hass.states.get(DEVICE_TRACKER_STATE)
self.assertEqual(state.attributes.get('gps_accuracy'), accuracy)
def assert_tracker_state(self, location):
"""Test the assertion of a tracker state."""
state = self.hass.states.get(REGION_TRACKER_STATE)
@ -312,7 +319,7 @@ class TestDeviceTrackerOwnTracks(unittest.TestCase):
state = self.hass.states.get(REGION_TRACKER_STATE)
self.assertEqual(state.attributes.get('gps_accuracy'), accuracy)
def test_location_invalid_devid(self):
def test_location_invalid_devid(self): # pylint: disable=invalid-name
"""Test the update of a location."""
self.send_message('owntracks/paulus/nexus-5x', LOCATION_MESSAGE)
@ -588,7 +595,7 @@ class TestDeviceTrackerOwnTracks(unittest.TestCase):
exit_message = REGION_LEAVE_MESSAGE.copy()
exit_message['desc'] = IBEACON_DEVICE
for i in range(0, 20):
for _ in range(0, 20):
fire_mqtt_message(
self.hass, EVENT_TOPIC, json.dumps(enter_message))
fire_mqtt_message(
@ -637,12 +644,16 @@ class TestDeviceTrackerOwnTracks(unittest.TestCase):
def test_waypoint_import_no_whitelist(self):
"""Test import of list of waypoints with no whitelist set."""
def mock_see(**kwargs):
"""Fake see method for owntracks."""
return
test_config = {
CONF_PLATFORM: 'owntracks',
CONF_MAX_GPS_ACCURACY: 200,
CONF_WAYPOINT_IMPORT: True
}
owntracks.setup_scanner(self.hass, test_config, self.mock_see)
owntracks.setup_scanner(self.hass, test_config, mock_see)
waypoints_message = WAYPOINTS_EXPORTED_MESSAGE.copy()
self.send_message(WAYPOINT_TOPIC_BLOCKED, waypoints_message)
# Check if it made it into states
@ -690,7 +701,18 @@ class TestDeviceTrackerOwnTracks(unittest.TestCase):
self.send_message(LOCATION_TOPIC, ENCRYPTED_LOCATION_MESSAGE)
self.assert_location_latitude(2.0)
def mock_cipher():
class TestDeviceTrackerOwnTrackConfigs(BaseMQTT):
"""Test the OwnTrack sensor."""
# pylint: disable=invalid-name
def setup_method(self, method):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
mock_mqtt_component(self.hass)
def mock_cipher(): # pylint: disable=no-method-argument
"""Return a dummy pickle-based cipher."""
def mock_decrypt(ciphertext, key):
"""Decrypt/unpickle."""
@ -705,11 +727,12 @@ class TestDeviceTrackerOwnTracks(unittest.TestCase):
mock_cipher)
def test_encrypted_payload(self):
"""Test encrypted payload."""
self.assertTrue(device_tracker.setup(self.hass, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_SECRET: SECRET_KEY,
}}))
with assert_setup_component(1, device_tracker.DOMAIN):
assert setup_component(self.hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_SECRET: SECRET_KEY,
}})
self.send_message(LOCATION_TOPIC, MOCK_ENCRYPTED_LOCATION_MESSAGE)
self.assert_location_latitude(2.0)
@ -717,24 +740,26 @@ class TestDeviceTrackerOwnTracks(unittest.TestCase):
mock_cipher)
def test_encrypted_payload_topic_key(self):
"""Test encrypted payload with a topic key."""
self.assertTrue(device_tracker.setup(self.hass, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_SECRET: {
LOCATION_TOPIC: SECRET_KEY,
}}}))
with assert_setup_component(1, device_tracker.DOMAIN):
assert setup_component(self.hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_SECRET: {
LOCATION_TOPIC: SECRET_KEY,
}}})
self.send_message(LOCATION_TOPIC, MOCK_ENCRYPTED_LOCATION_MESSAGE)
self.assert_location_latitude(2.0)
@patch('homeassistant.components.device_tracker.owntracks.get_cipher',
mock_cipher)
def test_encrypted_payload_no_key(self):
"""Test encrypted payload with no key."""
self.assertTrue(device_tracker.setup(self.hass, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
# key missing
}}))
"""Test encrypted payload with no key, ."""
with assert_setup_component(1, device_tracker.DOMAIN):
assert setup_component(self.hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
# key missing
}})
self.send_message(LOCATION_TOPIC, MOCK_ENCRYPTED_LOCATION_MESSAGE)
self.assert_location_latitude(None)
@ -742,11 +767,12 @@ class TestDeviceTrackerOwnTracks(unittest.TestCase):
mock_cipher)
def test_encrypted_payload_wrong_key(self):
"""Test encrypted payload with wrong key."""
self.assertTrue(device_tracker.setup(self.hass, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_SECRET: 'wrong key',
}}))
with assert_setup_component(1, device_tracker.DOMAIN):
assert setup_component(self.hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_SECRET: 'wrong key',
}})
self.send_message(LOCATION_TOPIC, MOCK_ENCRYPTED_LOCATION_MESSAGE)
self.assert_location_latitude(None)
@ -754,12 +780,13 @@ class TestDeviceTrackerOwnTracks(unittest.TestCase):
mock_cipher)
def test_encrypted_payload_wrong_topic_key(self):
"""Test encrypted payload with wrong topic key."""
self.assertTrue(device_tracker.setup(self.hass, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_SECRET: {
LOCATION_TOPIC: 'wrong key'
}}}))
with assert_setup_component(1, device_tracker.DOMAIN):
assert setup_component(self.hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_SECRET: {
LOCATION_TOPIC: 'wrong key'
}}})
self.send_message(LOCATION_TOPIC, MOCK_ENCRYPTED_LOCATION_MESSAGE)
self.assert_location_latitude(None)
@ -767,11 +794,12 @@ class TestDeviceTrackerOwnTracks(unittest.TestCase):
mock_cipher)
def test_encrypted_payload_no_topic_key(self):
"""Test encrypted payload with no topic key."""
self.assertTrue(device_tracker.setup(self.hass, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_SECRET: {
'owntracks/{}/{}'.format(USER, 'otherdevice'): 'foobar'
}}}))
with assert_setup_component(1, device_tracker.DOMAIN):
assert setup_component(self.hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_SECRET: {
'owntracks/{}/{}'.format(USER, 'otherdevice'): 'foobar'
}}})
self.send_message(LOCATION_TOPIC, MOCK_ENCRYPTED_LOCATION_MESSAGE)
self.assert_location_latitude(None)