jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-22 20:57:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

* Improved zone naming in waypoint import

Browse Source 
* Added more test coverage for owntracks and zone
This commit is contained in:
NMA 2016-08-28 13:18:30 +05:30
parent 2430acf3ad
commit 70fe7f747a
3 changed files with 78 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
homeassistant/components/device_tracker/owntracks.py
View File

@ -203,12 +203,13 @@ def setup_scanner(hass, config, see):
_LOGGER.info("Got %d waypoints from %s", len(wayps), topic)
for wayp in wayps:
name = wayp['desc']
pretty_name = parse_topic(topic, True)[1] + ' - ' + name
lat = wayp[WAYPOINT_LAT_KEY]
lon = wayp[WAYPOINT_LON_KEY]
rad = wayp['rad']
zone = zone_comp.Zone(hass, name, lat, lon, rad,
zone = zone_comp.Zone(hass, pretty_name, lat, lon, rad,
zone_comp.ICON_IMPORT, False, True)
zone_comp.add_zone(hass, name, zone)
zone_comp.add_zone(hass, pretty_name, zone)
def see_beacons(dev_id, kwargs_param):
"""Set active beacons to the current location."""
@ -236,11 +237,22 @@ def setup_scanner(hass, config, see):
return True
def parse_topic(topic, pretty=False):
"""Parse an MQTT topic owntracks/user/dev, return (user, dev) tuple."""
parts = topic.split('/')
dev_id_format = ''
if pretty:
dev_id_format = '{} {}'
else:
dev_id_format = '{}_{}'
dev_id = slugify(dev_id_format.format(parts[1], parts[2]))
host_name = parts[1]
return (host_name, dev_id)
def _parse_see_args(topic, data):
"""Parse the OwnTracks location parameters, into the format see expects."""
parts = topic.split('/')
dev_id = slugify('{}_{}'.format(parts[1], parts[2]))
host_name = parts[1]
(host_name, dev_id) = parse_topic(topic, False)
kwargs = {
'dev_id': dev_id,
'host_name': host_name,

7
homeassistant/components/zone.py
View File

@ -116,11 +116,10 @@ def add_zone(hass, name, zone, entities=None):
_entities = set()
else:
_entities = entities
zone_exists = hass.states.get('zone.' + str(name))
zone.entity_id = generate_entity_id(ENTITY_ID_FORMAT, name,
_entities)
zone_exists = hass.states.get(zone.entity_id)
if zone_exists is None:
zone.entity_id = generate_entity_id(ENTITY_ID_FORMAT, name,
_entities)
zone.update_ha_state()
_entities.add(zone.entity_id)
return zone

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

@ -136,6 +136,26 @@ WAYPOINTS_EXPORTED_MESSAGE = {
]
}
WAYPOINTS_UPDATED_MESSAGE = {
"_type": "waypoints",
"_creator": "test",
"waypoints": [
{
"_type": "waypoint",
"tst": 4,
"lat": 9,
"lon": 47,
"rad": 50,
"desc": "exp_wayp1"
},
]
}
WAYPOINT_ENTITY_NAMES = ['zone.greg_phone__exp_wayp1',
'zone.greg_phone__exp_wayp2',
'zone.ram_phone__exp_wayp1',
'zone.ram_phone__exp_wayp2']
REGION_ENTER_ZERO_MESSAGE = {
'lon': 1.0,
'event': 'enter',
@ -160,6 +180,9 @@ REGION_LEAVE_ZERO_MESSAGE = {
'lat': 20.0,
'_type': 'transition'}
BAD_JSON_PREFIX = '--$this is bad json#--'
BAD_JSON_SUFFIX = '** and it ends here ^^'
class TestDeviceTrackerOwnTracks(unittest.TestCase):
"""Test the OwnTrack sensor."""
@ -221,10 +244,14 @@ class TestDeviceTrackerOwnTracks(unittest.TestCase):
"""Fake see method for owntracks."""
return
def send_message(self, topic, message):
def send_message(self, topic, message, corrupt=False):
"""Test the sending of a message."""
fire_mqtt_message(
self.hass, topic, json.dumps(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.pool.block_till_done()
def assert_location_state(self, location):
@ -570,9 +597,9 @@ class TestDeviceTrackerOwnTracks(unittest.TestCase):
waypoints_message = WAYPOINTS_EXPORTED_MESSAGE.copy()
self.send_message(WAYPOINT_TOPIC, waypoints_message)
# Check if it made it into states
wayp = self.hass.states.get('zone.exp_wayp1')
wayp = self.hass.states.get(WAYPOINT_ENTITY_NAMES[0])
self.assertTrue(wayp is not None)
wayp = self.hass.states.get('zone.exp_wayp2')
wayp = self.hass.states.get(WAYPOINT_ENTITY_NAMES[1])
self.assertTrue(wayp is not None)
def test_waypoint_import_blacklist(self):
@ -580,9 +607,9 @@ class TestDeviceTrackerOwnTracks(unittest.TestCase):
waypoints_message = WAYPOINTS_EXPORTED_MESSAGE.copy()
self.send_message(WAYPOINT_TOPIC_BLOCKED, waypoints_message)
# Check if it made it into states
wayp = self.hass.states.get('zone.exp_wayp1')
wayp = self.hass.states.get(WAYPOINT_ENTITY_NAMES[2])
self.assertTrue(wayp is None)
wayp = self.hass.states.get('zone.exp_wayp2')
wayp = self.hass.states.get(WAYPOINT_ENTITY_NAMES[3])
self.assertTrue(wayp is None)
def test_waypoint_import_no_whitelist(self):
@ -596,7 +623,29 @@ class TestDeviceTrackerOwnTracks(unittest.TestCase):
waypoints_message = WAYPOINTS_EXPORTED_MESSAGE.copy()
self.send_message(WAYPOINT_TOPIC_BLOCKED, waypoints_message)
# Check if it made it into states
wayp = self.hass.states.get('zone.exp_wayp1')
wayp = self.hass.states.get(WAYPOINT_ENTITY_NAMES[2])
self.assertTrue(wayp is not None)
wayp = self.hass.states.get('zone.exp_wayp2')
wayp = self.hass.states.get(WAYPOINT_ENTITY_NAMES[3])
self.assertTrue(wayp is not None)
def test_waypoint_import_bad_json(self):
"""Test importing a bad JSON payload."""
waypoints_message = WAYPOINTS_EXPORTED_MESSAGE.copy()
self.send_message(WAYPOINT_TOPIC, waypoints_message, True)
# Check if it made it into states
wayp = self.hass.states.get(WAYPOINT_ENTITY_NAMES[2])
self.assertTrue(wayp is None)
wayp = self.hass.states.get(WAYPOINT_ENTITY_NAMES[3])
self.assertTrue(wayp is None)
def test_waypoint_import_existing(self):
"""Test importing a zone that exists."""
waypoints_message = WAYPOINTS_EXPORTED_MESSAGE.copy()
self.send_message(WAYPOINT_TOPIC, waypoints_message)
# Get the first waypoint exported
wayp = self.hass.states.get(WAYPOINT_ENTITY_NAMES[0])
# Send an update
waypoints_message = WAYPOINTS_UPDATED_MESSAGE.copy()
self.send_message(WAYPOINT_TOPIC, waypoints_message)
new_wayp = self.hass.states.get(WAYPOINT_ENTITY_NAMES[0])
self.assertTrue(wayp == new_wayp)