diff --git a/homeassistant/components/device_tracker/owntracks.py b/homeassistant/components/device_tracker/owntracks.py index 0d7e17307d4..77241e1a8ab 100644 --- a/homeassistant/components/device_tracker/owntracks.py +++ b/homeassistant/components/device_tracker/owntracks.py @@ -199,7 +199,7 @@ class OwnTracksContext: self.async_see = async_see self.secret = secret self.max_gps_accuracy = max_gps_accuracy - self.mobile_beacons_active = defaultdict(list) + self.mobile_beacons_active = defaultdict(set) self.regions_entered = defaultdict(list) self.import_waypoints = import_waypoints self.waypoint_whitelist = waypoint_whitelist @@ -234,10 +234,25 @@ class OwnTracksContext: return True @asyncio.coroutine - def async_see_beacons(self, dev_id, kwargs_param): + def async_see_beacons(self, hass, dev_id, kwargs_param): """Set active beacons to the current location.""" kwargs = kwargs_param.copy() + + # Mobile beacons should always be set to the location of the + # tracking device. I get the device state and make the necessary + # changes to kwargs. + device_tracker_state = hass.states.get( + "device_tracker.{}".format(dev_id)) + + if device_tracker_state is not None: + acc = device_tracker_state.attributes.get("gps_accuracy") + lat = device_tracker_state.attributes.get("latitude") + lon = device_tracker_state.attributes.get("longitude") + kwargs['gps_accuracy'] = acc + kwargs['gps'] = (lat, lon) + # the battery state applies to the tracking device, not the beacon + # kwargs location is the beacon's configured lat/lon kwargs.pop('battery', None) for beacon in self.mobile_beacons_active[dev_id]: kwargs['dev_id'] = "{}_{}".format(BEACON_DEV_ID, beacon) @@ -261,7 +276,7 @@ def async_handle_location_message(hass, context, message): return yield from context.async_see(**kwargs) - yield from context.async_see_beacons(dev_id, kwargs) + yield from context.async_see_beacons(hass, dev_id, kwargs) @asyncio.coroutine @@ -271,11 +286,15 @@ def _async_transition_message_enter(hass, context, message, location): dev_id, kwargs = _parse_see_args(message) if zone is None and message.get('t') == 'b': - # Not a HA zone, and a beacon so assume mobile + # Not a HA zone, and a beacon so mobile beacon. + # kwargs will contain the lat/lon of the beacon + # which is not where the beacon actually is + # and is probably set to 0/0 beacons = context.mobile_beacons_active[dev_id] if location not in beacons: - beacons.append(location) + beacons.add(location) _LOGGER.info("Added beacon %s", location) + yield from context.async_see_beacons(hass, dev_id, kwargs) else: # Normal region regions = context.regions_entered[dev_id] @@ -283,9 +302,8 @@ def _async_transition_message_enter(hass, context, message, location): regions.append(location) _LOGGER.info("Enter region %s", location) _set_gps_from_zone(kwargs, location, zone) - - yield from context.async_see(**kwargs) - yield from context.async_see_beacons(dev_id, kwargs) + yield from context.async_see(**kwargs) + yield from context.async_see_beacons(hass, dev_id, kwargs) @asyncio.coroutine @@ -297,30 +315,29 @@ def _async_transition_message_leave(hass, context, message, location): if location in regions: regions.remove(location) - new_region = regions[-1] if regions else None - - if new_region: - # Exit to previous region - zone = hass.states.get( - "zone.{}".format(slugify(new_region))) - _set_gps_from_zone(kwargs, new_region, zone) - _LOGGER.info("Exit to %s", new_region) - yield from context.async_see(**kwargs) - yield from context.async_see_beacons(dev_id, kwargs) - return - + beacons = context.mobile_beacons_active[dev_id] + if location in beacons: + beacons.remove(location) + _LOGGER.info("Remove beacon %s", location) + yield from context.async_see_beacons(hass, dev_id, kwargs) else: + new_region = regions[-1] if regions else None + if new_region: + # Exit to previous region + zone = hass.states.get( + "zone.{}".format(slugify(new_region))) + _set_gps_from_zone(kwargs, new_region, zone) + _LOGGER.info("Exit to %s", new_region) + yield from context.async_see(**kwargs) + yield from context.async_see_beacons(hass, dev_id, kwargs) + return + _LOGGER.info("Exit to GPS") # Check for GPS accuracy if context.async_valid_accuracy(message): yield from context.async_see(**kwargs) - yield from context.async_see_beacons(dev_id, kwargs) - - beacons = context.mobile_beacons_active[dev_id] - if location in beacons: - beacons.remove(location) - _LOGGER.info("Remove beacon %s", location) + yield from context.async_see_beacons(hass, dev_id, kwargs) @HANDLERS.register('transition') diff --git a/tests/components/device_tracker/test_owntracks.py b/tests/components/device_tracker/test_owntracks.py index eb163fdcbdf..a06adcb286a 100644 --- a/tests/components/device_tracker/test_owntracks.py +++ b/tests/components/device_tracker/test_owntracks.py @@ -26,95 +26,174 @@ WAYPOINT_TOPIC_BLOCKED = 'owntracks/{}/{}/waypoints'.format( DEVICE_TRACKER_STATE = 'device_tracker.{}_{}'.format(USER, DEVICE) IBEACON_DEVICE = 'keys' -REGION_TRACKER_STATE = 'device_tracker.beacon_{}'.format(IBEACON_DEVICE) +MOBILE_BEACON_FMT = 'device_tracker.beacon_{}' CONF_MAX_GPS_ACCURACY = 'max_gps_accuracy' CONF_WAYPOINT_IMPORT = owntracks.CONF_WAYPOINT_IMPORT CONF_WAYPOINT_WHITELIST = owntracks.CONF_WAYPOINT_WHITELIST CONF_SECRET = owntracks.CONF_SECRET -LOCATION_MESSAGE = { - 'batt': 92, - 'cog': 248, - 'tid': 'user', - 'lon': 1.0, - 't': 'u', - 'alt': 27, +TEST_ZONE_LAT = 45.0 +TEST_ZONE_LON = 90.0 +TEST_ZONE_DEG_PER_M = 0.0000127 +FIVE_M = TEST_ZONE_DEG_PER_M * 5.0 + + +# Home Assistant Zones +INNER_ZONE = { + 'name': 'zone', + 'latitude': TEST_ZONE_LAT+0.1, + 'longitude': TEST_ZONE_LON+0.1, + 'radius': 50 +} + +OUTER_ZONE = { + 'name': 'zone', + 'latitude': TEST_ZONE_LAT, + 'longitude': TEST_ZONE_LON, + 'radius': 100000 +} + + +def build_message(test_params, default_params): + """Build a test message from overrides and another message.""" + new_params = default_params.copy() + new_params.update(test_params) + return new_params + + +# Default message parameters +DEFAULT_LOCATION_MESSAGE = { + '_type': 'location', + 'lon': OUTER_ZONE['longitude'], + 'lat': OUTER_ZONE['latitude'], 'acc': 60, - 'p': 101.3977584838867, - 'vac': 4, - 'lat': 2.0, - '_type': 'location', - 'tst': 1, - 'vel': 0} - -LOCATION_MESSAGE_INACCURATE = { + 'tid': 'user', + 't': 'u', 'batt': 92, 'cog': 248, - 'tid': 'user', - 'lon': 2.0, - 't': 'u', 'alt': 27, - 'acc': 2000, 'p': 101.3977584838867, 'vac': 4, - 'lat': 6.0, - '_type': 'location', 'tst': 1, - 'vel': 0} + 'vel': 0 +} -LOCATION_MESSAGE_ZERO_ACCURACY = { - 'batt': 92, - 'cog': 248, - 'tid': 'user', - 'lon': 2.0, - 't': 'u', - 'alt': 27, - 'acc': 0, - 'p': 101.3977584838867, - 'vac': 4, - 'lat': 6.0, - '_type': 'location', - 'tst': 1, - 'vel': 0} - -REGION_ENTER_MESSAGE = { - 'lon': 1.0, +# Owntracks will publish a transition when crossing +# a circular region boundary. +ZONE_EDGE = TEST_ZONE_DEG_PER_M * INNER_ZONE['radius'] +DEFAULT_TRANSITION_MESSAGE = { + '_type': 'transition', + 't': 'c', + 'lon': INNER_ZONE['longitude'], + 'lat': INNER_ZONE['latitude'] - ZONE_EDGE, + 'acc': 60, 'event': 'enter', 'tid': 'user', 'desc': 'inner', 'wtst': 1, + 'tst': 2 +} + +# iBeacons that are named the same as an HA zone +# are used to trigger enter and leave updates +# for that zone. In this case the "inner" zone. +# +# iBeacons that do not share an HA zone name +# are treated as mobile tracking devices for +# objects which can't track themselves e.g. keys. +# +# iBeacons are typically configured with the +# default lat/lon 0.0/0.0 and have acc 0.0 but +# regardless the reported location is not trusted. +# +# Owntracks will send both a location message +# for the device and an 'event' message for +# the beacon transition. +DEFAULT_BEACON_TRANSITION_MESSAGE = { + '_type': 'transition', 't': 'b', - 'acc': 60, - 'tst': 2, - 'lat': 2.0, - '_type': 'transition'} - - -REGION_LEAVE_MESSAGE = { - 'lon': 1.0, - 'event': 'leave', + 'lon': 0.0, + 'lat': 0.0, + 'acc': 0.0, + 'event': 'enter', 'tid': 'user', 'desc': 'inner', 'wtst': 1, - 't': 'b', - 'acc': 60, - 'tst': 2, - 'lat': 2.0, - '_type': 'transition'} + 'tst': 2 +} -REGION_LEAVE_INACCURATE_MESSAGE = { - 'lon': 10.0, - 'event': 'leave', - 'tid': 'user', - 'desc': 'inner', - 'wtst': 1, - 't': 'b', - 'acc': 2000, - 'tst': 2, - 'lat': 20.0, - '_type': 'transition'} +# Location messages +LOCATION_MESSAGE = DEFAULT_LOCATION_MESSAGE +LOCATION_MESSAGE_INACCURATE = build_message( + {'lat': INNER_ZONE['latitude'] - ZONE_EDGE, + 'lon': INNER_ZONE['longitude'] - ZONE_EDGE, + 'acc': 2000}, + LOCATION_MESSAGE) + +LOCATION_MESSAGE_ZERO_ACCURACY = build_message( + {'lat': INNER_ZONE['latitude'] - ZONE_EDGE, + 'lon': INNER_ZONE['longitude'] - ZONE_EDGE, + 'acc': 0}, + LOCATION_MESSAGE) + +LOCATION_MESSAGE_NOT_HOME = build_message( + {'lat': OUTER_ZONE['latitude'] - 2.0, + 'lon': INNER_ZONE['longitude'] - 2.0, + 'acc': 100}, + LOCATION_MESSAGE) + +# Region GPS messages +REGION_GPS_ENTER_MESSAGE = DEFAULT_TRANSITION_MESSAGE + +REGION_GPS_LEAVE_MESSAGE = build_message( + {'lon': INNER_ZONE['longitude'] - ZONE_EDGE * 10, + 'lat': INNER_ZONE['latitude'] - ZONE_EDGE * 10, + 'event': 'leave'}, + DEFAULT_TRANSITION_MESSAGE) + +REGION_GPS_ENTER_MESSAGE_INACCURATE = build_message( + {'acc': 2000}, + REGION_GPS_ENTER_MESSAGE) + +REGION_GPS_LEAVE_MESSAGE_INACCURATE = build_message( + {'acc': 2000}, + REGION_GPS_LEAVE_MESSAGE) + +REGION_GPS_ENTER_MESSAGE_ZERO = build_message( + {'acc': 0}, + REGION_GPS_ENTER_MESSAGE) + +REGION_GPS_LEAVE_MESSAGE_ZERO = build_message( + {'acc': 0}, + REGION_GPS_LEAVE_MESSAGE) + +REGION_GPS_LEAVE_MESSAGE_OUTER = build_message( + {'lon': OUTER_ZONE['longitude'] - 2.0, + 'lat': OUTER_ZONE['latitude'] - 2.0, + 'desc': 'outer', + 'event': 'leave'}, + DEFAULT_TRANSITION_MESSAGE) + +# Region Beacon messages +REGION_BEACON_ENTER_MESSAGE = DEFAULT_BEACON_TRANSITION_MESSAGE + +REGION_BEACON_LEAVE_MESSAGE = build_message( + {'event': 'leave'}, + DEFAULT_BEACON_TRANSITION_MESSAGE) + +# Mobile Beacon messages +MOBILE_BEACON_ENTER_EVENT_MESSAGE = build_message( + {'desc': IBEACON_DEVICE}, + DEFAULT_BEACON_TRANSITION_MESSAGE) + +MOBILE_BEACON_LEAVE_EVENT_MESSAGE = build_message( + {'desc': IBEACON_DEVICE, + 'event': 'leave'}, + DEFAULT_BEACON_TRANSITION_MESSAGE) + +# Waypoint messages WAYPOINTS_EXPORTED_MESSAGE = { "_type": "waypoints", "_creator": "test", @@ -160,54 +239,9 @@ WAYPOINT_ENTITY_NAMES = [ 'zone.ram_phone__exp_wayp2', ] -REGION_ENTER_ZERO_MESSAGE = { - 'lon': 1.0, - 'event': 'enter', - 'tid': 'user', - 'desc': 'inner', - 'wtst': 1, - 't': 'b', - 'acc': 0, - 'tst': 2, - 'lat': 2.0, - '_type': 'transition'} - -REGION_LEAVE_ZERO_MESSAGE = { - 'lon': 10.0, - 'event': 'leave', - 'tid': 'user', - 'desc': 'inner', - 'wtst': 1, - 't': 'b', - 'acc': 0, - 'tst': 2, - 'lat': 20.0, - '_type': 'transition'} - BAD_JSON_PREFIX = '--$this is bad json#--' BAD_JSON_SUFFIX = '** and it ends here ^^' -TEST_SECRET_KEY = 's3cretkey' -ENCRYPTED_LOCATION_MESSAGE = { - # Encrypted version of LOCATION_MESSAGE using libsodium and TEST_SECRET_KEY - '_type': 'encrypted', - 'data': ('qm1A83I6TVFRmH5343xy+cbex8jBBxDFkHRuJhELVKVRA/DgXcyKtghw' - '9pOw75Lo4gHcyy2wV5CmkjrpKEBR7Qhye4AR0y7hOvlx6U/a3GuY1+W8' - 'I4smrLkwMvGgBOzXSNdVTzbFTHDvG3gRRaNHFkt2+5MsbH2Dd6CXmpzq' - 'DIfSN7QzwOevuvNIElii5MlFxI6ZnYIDYA/ZdnAXHEVsNIbyT2N0CXt3' - 'fTPzgGtFzsufx40EEUkC06J7QTJl7lLG6qaLW1cCWp86Vp0eL3vtZ6xq') -} - -MOCK_ENCRYPTED_LOCATION_MESSAGE = { - # Mock-encrypted version of LOCATION_MESSAGE using pickle - '_type': 'encrypted', - 'data': ('gANDCXMzY3JldGtleXEAQ6p7ImxvbiI6IDEuMCwgInQiOiAidSIsICJi' - 'YXR0IjogOTIsICJhY2MiOiA2MCwgInZlbCI6IDAsICJfdHlwZSI6ICJs' - 'b2NhdGlvbiIsICJ2YWMiOiA0LCAicCI6IDEwMS4zOTc3NTg0ODM4ODY3' - 'LCAidHN0IjogMSwgImxhdCI6IDIuMCwgImFsdCI6IDI3LCAiY29nIjog' - 'MjQ4LCAidGlkIjogInVzZXIifXEBhnECLg==') -} - class BaseMQTT(unittest.TestCase): """Base MQTT assert functions.""" @@ -282,58 +316,46 @@ class TestDeviceTrackerOwnTracks(BaseMQTT): }}) self.hass.states.set( - 'zone.inner', 'zoning', - { - 'name': 'zone', - 'latitude': 2.1, - 'longitude': 1.1, - 'radius': 10 - }) + 'zone.inner', 'zoning', INNER_ZONE) self.hass.states.set( - 'zone.inner_2', 'zoning', - { - 'name': 'zone', - 'latitude': 2.1, - 'longitude': 1.1, - 'radius': 10 - }) + 'zone.inner_2', 'zoning', INNER_ZONE) self.hass.states.set( - 'zone.outer', 'zoning', - { - 'name': 'zone', - 'latitude': 2.0, - 'longitude': 1.0, - 'radius': 100000 - }) + 'zone.outer', 'zoning', OUTER_ZONE) - # Clear state between teste + # Clear state between tests + # NB: state "None" is not a state that is created by Device + # so when we compare state to None in the tests this + # is really checking that it is still in its original + # test case state. See Device.async_update. self.hass.states.set(DEVICE_TRACKER_STATE, None) def teardown_method(self, _): """Stop everything that was started.""" self.hass.stop() - def assert_tracker_state(self, location): - """Test the assertion of a tracker state.""" - state = self.hass.states.get(REGION_TRACKER_STATE) + def assert_mobile_tracker_state(self, location, beacon=IBEACON_DEVICE): + """Test the assertion of a mobile beacon tracker state.""" + dev_id = MOBILE_BEACON_FMT.format(beacon) + state = self.hass.states.get(dev_id) self.assertEqual(state.state, location) - def assert_tracker_latitude(self, latitude): - """Test the assertion of a tracker latitude.""" - state = self.hass.states.get(REGION_TRACKER_STATE) + def assert_mobile_tracker_latitude(self, latitude, beacon=IBEACON_DEVICE): + """Test the assertion of a mobile beacon tracker latitude.""" + dev_id = MOBILE_BEACON_FMT.format(beacon) + state = self.hass.states.get(dev_id) self.assertEqual(state.attributes.get('latitude'), latitude) - def assert_tracker_accuracy(self, accuracy): - """Test the assertion of a tracker accuracy.""" - state = self.hass.states.get(REGION_TRACKER_STATE) + def assert_mobile_tracker_accuracy(self, accuracy, beacon=IBEACON_DEVICE): + """Test the assertion of a mobile beacon tracker accuracy.""" + dev_id = MOBILE_BEACON_FMT.format(beacon) + state = self.hass.states.get(dev_id) self.assertEqual(state.attributes.get('gps_accuracy'), accuracy) 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) - state = self.hass.states.get('device_tracker.paulus_nexus5x') assert state.state == 'outer' @@ -341,8 +363,8 @@ class TestDeviceTrackerOwnTracks(BaseMQTT): """Test the update of a location.""" self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE) - self.assert_location_latitude(2.0) - self.assert_location_accuracy(60.0) + self.assert_location_latitude(LOCATION_MESSAGE['lat']) + self.assert_location_accuracy(LOCATION_MESSAGE['acc']) self.assert_location_state('outer') def test_location_inaccurate_gps(self): @@ -350,288 +372,686 @@ class TestDeviceTrackerOwnTracks(BaseMQTT): self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE) self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE_INACCURATE) - self.assert_location_latitude(2.0) - self.assert_location_longitude(1.0) + # Ignored inaccurate GPS. Location remains at previous. + self.assert_location_latitude(LOCATION_MESSAGE['lat']) + self.assert_location_longitude(LOCATION_MESSAGE['lon']) def test_location_zero_accuracy_gps(self): """Ignore the location for zero accuracy GPS information.""" self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE) self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE_ZERO_ACCURACY) - self.assert_location_latitude(2.0) - self.assert_location_longitude(1.0) + # Ignored inaccurate GPS. Location remains at previous. + self.assert_location_latitude(LOCATION_MESSAGE['lat']) + self.assert_location_longitude(LOCATION_MESSAGE['lon']) - def test_event_entry_exit(self): + # ------------------------------------------------------------------------ + # GPS based event entry / exit testing + + def test_event_gps_entry_exit(self): """Test the entry event.""" - self.send_message(EVENT_TOPIC, REGION_ENTER_MESSAGE) + # Entering the owntrack circular region named "inner" + self.send_message(EVENT_TOPIC, REGION_GPS_ENTER_MESSAGE) # Enter uses the zone's gps co-ords - self.assert_location_latitude(2.1) - self.assert_location_accuracy(10.0) + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_location_accuracy(INNER_ZONE['radius']) self.assert_location_state('inner') self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE) # Updates ignored when in a zone - self.assert_location_latitude(2.1) - self.assert_location_accuracy(10.0) + # note that LOCATION_MESSAGE is actually pretty far + # from INNER_ZONE and has good accuracy. I haven't + # received a transition message though so I'm still + # asssociated with the inner zone regardless of GPS. + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_location_accuracy(INNER_ZONE['radius']) self.assert_location_state('inner') - self.send_message(EVENT_TOPIC, REGION_LEAVE_MESSAGE) + self.send_message(EVENT_TOPIC, REGION_GPS_LEAVE_MESSAGE) # Exit switches back to GPS - self.assert_location_latitude(2.0) - self.assert_location_accuracy(60.0) + self.assert_location_latitude(REGION_GPS_LEAVE_MESSAGE['lat']) + self.assert_location_accuracy(REGION_GPS_LEAVE_MESSAGE['acc']) self.assert_location_state('outer') # Left clean zone state self.assertFalse(self.context.regions_entered[USER]) - def test_event_with_spaces(self): + self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE) + + # Now sending a location update moves me again. + self.assert_location_latitude(LOCATION_MESSAGE['lat']) + self.assert_location_accuracy(LOCATION_MESSAGE['acc']) + + def test_event_gps_with_spaces(self): """Test the entry event.""" - message = REGION_ENTER_MESSAGE.copy() - message['desc'] = "inner 2" + message = build_message({'desc': "inner 2"}, + REGION_GPS_ENTER_MESSAGE) self.send_message(EVENT_TOPIC, message) self.assert_location_state('inner 2') - message = REGION_LEAVE_MESSAGE.copy() - message['desc'] = "inner 2" + message = build_message({'desc': "inner 2"}, + REGION_GPS_LEAVE_MESSAGE) self.send_message(EVENT_TOPIC, message) # Left clean zone state self.assertFalse(self.context.regions_entered[USER]) - def test_event_entry_exit_inaccurate(self): - """Test the event for inaccurate exit.""" - self.send_message(EVENT_TOPIC, REGION_ENTER_MESSAGE) + def test_event_gps_entry_inaccurate(self): + """Test the event for inaccurate entry.""" + # Set location to the outer zone. + self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE) - # Enter uses the zone's gps co-ords - self.assert_location_latitude(2.1) - self.assert_location_accuracy(10.0) + self.send_message(EVENT_TOPIC, REGION_GPS_ENTER_MESSAGE_INACCURATE) + + # I enter the zone even though the message GPS was inaccurate. + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_location_accuracy(INNER_ZONE['radius']) self.assert_location_state('inner') - self.send_message(EVENT_TOPIC, REGION_LEAVE_INACCURATE_MESSAGE) + def test_event_gps_entry_exit_inaccurate(self): + """Test the event for inaccurate exit.""" + self.send_message(EVENT_TOPIC, REGION_GPS_ENTER_MESSAGE) + + # Enter uses the zone's gps co-ords + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_location_accuracy(INNER_ZONE['radius']) + self.assert_location_state('inner') + + self.send_message(EVENT_TOPIC, REGION_GPS_LEAVE_MESSAGE_INACCURATE) # Exit doesn't use inaccurate gps - self.assert_location_latitude(2.1) - self.assert_location_accuracy(10.0) + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_location_accuracy(INNER_ZONE['radius']) self.assert_location_state('inner') # But does exit region correctly self.assertFalse(self.context.regions_entered[USER]) - def test_event_entry_exit_zero_accuracy(self): + def test_event_gps_entry_exit_zero_accuracy(self): """Test entry/exit events with accuracy zero.""" - self.send_message(EVENT_TOPIC, REGION_ENTER_ZERO_MESSAGE) + self.send_message(EVENT_TOPIC, REGION_GPS_ENTER_MESSAGE_ZERO) # Enter uses the zone's gps co-ords - self.assert_location_latitude(2.1) - self.assert_location_accuracy(10.0) + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_location_accuracy(INNER_ZONE['radius']) self.assert_location_state('inner') - self.send_message(EVENT_TOPIC, REGION_LEAVE_ZERO_MESSAGE) + self.send_message(EVENT_TOPIC, REGION_GPS_LEAVE_MESSAGE_ZERO) # Exit doesn't use zero gps - self.assert_location_latitude(2.1) - self.assert_location_accuracy(10.0) + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_location_accuracy(INNER_ZONE['radius']) self.assert_location_state('inner') # But does exit region correctly self.assertFalse(self.context.regions_entered[USER]) - def test_event_exit_outside_zone_sets_away(self): + def test_event_gps_exit_outside_zone_sets_away(self): """Test the event for exit zone.""" - self.send_message(EVENT_TOPIC, REGION_ENTER_MESSAGE) + self.send_message(EVENT_TOPIC, REGION_GPS_ENTER_MESSAGE) self.assert_location_state('inner') # Exit message far away GPS location - message = REGION_LEAVE_MESSAGE.copy() - message['lon'] = 90.1 - message['lat'] = 90.1 + message = build_message( + {'lon': 90.0, + 'lat': 90.0}, + REGION_GPS_LEAVE_MESSAGE) self.send_message(EVENT_TOPIC, message) # Exit forces zone change to away self.assert_location_state(STATE_NOT_HOME) - def test_event_entry_exit_right_order(self): + def test_event_gps_entry_exit_right_order(self): """Test the event for ordering.""" # Enter inner zone - self.send_message(EVENT_TOPIC, REGION_ENTER_MESSAGE) - + # Set location to the outer zone. + self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE) + self.send_message(EVENT_TOPIC, REGION_GPS_ENTER_MESSAGE) self.assert_location_state('inner') - self.assert_location_latitude(2.1) - self.assert_location_accuracy(10.0) # Enter inner2 zone - message = REGION_ENTER_MESSAGE.copy() - message['desc'] = "inner_2" + message = build_message( + {'desc': "inner_2"}, + REGION_GPS_ENTER_MESSAGE) self.send_message(EVENT_TOPIC, message) self.assert_location_state('inner_2') - self.assert_location_latitude(2.1) - self.assert_location_accuracy(10.0) # Exit inner_2 - should be in 'inner' - message = REGION_LEAVE_MESSAGE.copy() - message['desc'] = "inner_2" + message = build_message( + {'desc': "inner_2"}, + REGION_GPS_LEAVE_MESSAGE) self.send_message(EVENT_TOPIC, message) self.assert_location_state('inner') - self.assert_location_latitude(2.1) - self.assert_location_accuracy(10.0) # Exit inner - should be in 'outer' - self.send_message(EVENT_TOPIC, REGION_LEAVE_MESSAGE) + self.send_message(EVENT_TOPIC, REGION_GPS_LEAVE_MESSAGE) + self.assert_location_latitude(REGION_GPS_LEAVE_MESSAGE['lat']) + self.assert_location_accuracy(REGION_GPS_LEAVE_MESSAGE['acc']) self.assert_location_state('outer') - self.assert_location_latitude(2.0) - self.assert_location_accuracy(60.0) - def test_event_entry_exit_wrong_order(self): + def test_event_gps_entry_exit_wrong_order(self): """Test the event for wrong order.""" # Enter inner zone - self.send_message(EVENT_TOPIC, REGION_ENTER_MESSAGE) + self.send_message(EVENT_TOPIC, REGION_GPS_ENTER_MESSAGE) self.assert_location_state('inner') # Enter inner2 zone - message = REGION_ENTER_MESSAGE.copy() - message['desc'] = "inner_2" + message = build_message( + {'desc': "inner_2"}, + REGION_GPS_ENTER_MESSAGE) self.send_message(EVENT_TOPIC, message) self.assert_location_state('inner_2') # Exit inner - should still be in 'inner_2' - self.send_message(EVENT_TOPIC, REGION_LEAVE_MESSAGE) + self.send_message(EVENT_TOPIC, REGION_GPS_LEAVE_MESSAGE) self.assert_location_state('inner_2') # Exit inner_2 - should be in 'outer' - message = REGION_LEAVE_MESSAGE.copy() - message['desc'] = "inner_2" + message = build_message( + {'desc': "inner_2"}, + REGION_GPS_LEAVE_MESSAGE) self.send_message(EVENT_TOPIC, message) + self.assert_location_latitude(REGION_GPS_LEAVE_MESSAGE['lat']) + self.assert_location_accuracy(REGION_GPS_LEAVE_MESSAGE['acc']) self.assert_location_state('outer') - def test_event_entry_unknown_zone(self): + def test_event_gps_entry_unknown_zone(self): """Test the event for unknown zone.""" # Just treat as location update - message = REGION_ENTER_MESSAGE.copy() - message['desc'] = "unknown" + message = build_message( + {'desc': "unknown"}, + REGION_GPS_ENTER_MESSAGE) self.send_message(EVENT_TOPIC, message) - self.assert_location_latitude(2.0) - self.assert_location_state('outer') + self.assert_location_latitude(REGION_GPS_ENTER_MESSAGE['lat']) + self.assert_location_state('inner') - def test_event_exit_unknown_zone(self): + def test_event_gps_exit_unknown_zone(self): """Test the event for unknown zone.""" # Just treat as location update - message = REGION_LEAVE_MESSAGE.copy() - message['desc'] = "unknown" + message = build_message( + {'desc': "unknown"}, + REGION_GPS_LEAVE_MESSAGE) self.send_message(EVENT_TOPIC, message) - self.assert_location_latitude(2.0) + self.assert_location_latitude(REGION_GPS_LEAVE_MESSAGE['lat']) self.assert_location_state('outer') def test_event_entry_zone_loading_dash(self): """Test the event for zone landing.""" # Make sure the leading - is ignored # Ownracks uses this to switch on hold - message = REGION_ENTER_MESSAGE.copy() - message['desc'] = "-inner" - self.send_message(EVENT_TOPIC, REGION_ENTER_MESSAGE) - + message = build_message( + {'desc': "-inner"}, + REGION_GPS_ENTER_MESSAGE) + self.send_message(EVENT_TOPIC, message) self.assert_location_state('inner') + # Region Beacon based event entry / exit testing + + def test_event_region_entry_exit(self): + """Test the entry event.""" + # Seeing a beacon named "inner" + self.send_message(EVENT_TOPIC, REGION_BEACON_ENTER_MESSAGE) + + # Enter uses the zone's gps co-ords + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_location_accuracy(INNER_ZONE['radius']) + self.assert_location_state('inner') + + self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE) + + # Updates ignored when in a zone + # note that LOCATION_MESSAGE is actually pretty far + # from INNER_ZONE and has good accuracy. I haven't + # received a transition message though so I'm still + # asssociated with the inner zone regardless of GPS. + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_location_accuracy(INNER_ZONE['radius']) + self.assert_location_state('inner') + + self.send_message(EVENT_TOPIC, REGION_BEACON_LEAVE_MESSAGE) + + # Exit switches back to GPS but the beacon has no coords + # so I am still located at the center of the inner region + # until I receive a location update. + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_location_accuracy(INNER_ZONE['radius']) + self.assert_location_state('inner') + + # Left clean zone state + self.assertFalse(self.context.regions_entered[USER]) + + # Now sending a location update moves me again. + self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE) + self.assert_location_latitude(LOCATION_MESSAGE['lat']) + self.assert_location_accuracy(LOCATION_MESSAGE['acc']) + + def test_event_region_with_spaces(self): + """Test the entry event.""" + message = build_message({'desc': "inner 2"}, + REGION_BEACON_ENTER_MESSAGE) + self.send_message(EVENT_TOPIC, message) + self.assert_location_state('inner 2') + + message = build_message({'desc': "inner 2"}, + REGION_BEACON_LEAVE_MESSAGE) + self.send_message(EVENT_TOPIC, message) + + # Left clean zone state + self.assertFalse(self.context.regions_entered[USER]) + + def test_event_region_entry_exit_right_order(self): + """Test the event for ordering.""" + # Enter inner zone + # Set location to the outer zone. + self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE) + + # See 'inner' region beacon + self.send_message(EVENT_TOPIC, REGION_BEACON_ENTER_MESSAGE) + self.assert_location_state('inner') + + # See 'inner_2' region beacon + message = build_message( + {'desc': "inner_2"}, + REGION_BEACON_ENTER_MESSAGE) + self.send_message(EVENT_TOPIC, message) + self.assert_location_state('inner_2') + + # Exit inner_2 - should be in 'inner' + message = build_message( + {'desc': "inner_2"}, + REGION_BEACON_LEAVE_MESSAGE) + self.send_message(EVENT_TOPIC, message) + self.assert_location_state('inner') + + # Exit inner - should be in 'outer' + self.send_message(EVENT_TOPIC, REGION_BEACON_LEAVE_MESSAGE) + + # I have not had an actual location update yet and my + # coordinates are set to the center of the last region I + # entered which puts me in the inner zone. + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_location_accuracy(INNER_ZONE['radius']) + self.assert_location_state('inner') + + def test_event_region_entry_exit_wrong_order(self): + """Test the event for wrong order.""" + # Enter inner zone + self.send_message(EVENT_TOPIC, REGION_BEACON_ENTER_MESSAGE) + self.assert_location_state('inner') + + # Enter inner2 zone + message = build_message( + {'desc': "inner_2"}, + REGION_BEACON_ENTER_MESSAGE) + self.send_message(EVENT_TOPIC, message) + self.assert_location_state('inner_2') + + # Exit inner - should still be in 'inner_2' + self.send_message(EVENT_TOPIC, REGION_BEACON_LEAVE_MESSAGE) + self.assert_location_state('inner_2') + + # Exit inner_2 - should be in 'outer' + message = build_message( + {'desc': "inner_2"}, + REGION_BEACON_LEAVE_MESSAGE) + self.send_message(EVENT_TOPIC, message) + + # I have not had an actual location update yet and my + # coordinates are set to the center of the last region I + # entered which puts me in the inner_2 zone. + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_location_accuracy(INNER_ZONE['radius']) + self.assert_location_state('inner_2') + + def test_event_beacon_unknown_zone_no_location(self): + """Test the event for unknown zone.""" + # A beacon which does not match a HA zone is the + # definition of a mobile beacon. In this case, "unknown" + # will be turned into device_tracker.beacon_unknown and + # that will be tracked at my current location. Except + # in this case my Device hasn't had a location message + # yet so it's in an odd state where it has state.state + # None and no GPS coords so set the beacon to. + + message = build_message( + {'desc': "unknown"}, + REGION_BEACON_ENTER_MESSAGE) + self.send_message(EVENT_TOPIC, message) + + # My current state is None because I haven't seen a + # location message or a GPS or Region # Beacon event + # message. None is the state the test harness set for + # the Device during test case setup. + self.assert_location_state('None') + + # home is the state of a Device constructed through + # the normal code path on it's first observation with + # the conditions I pass along. + self.assert_mobile_tracker_state('home', 'unknown') + + def test_event_beacon_unknown_zone(self): + """Test the event for unknown zone.""" + # A beacon which does not match a HA zone is the + # definition of a mobile beacon. In this case, "unknown" + # will be turned into device_tracker.beacon_unknown and + # that will be tracked at my current location. First I + # set my location so that my state is 'outer' + + self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE) + self.assert_location_state('outer') + + message = build_message( + {'desc': "unknown"}, + REGION_BEACON_ENTER_MESSAGE) + self.send_message(EVENT_TOPIC, message) + + # My state is still outer and now the unknown beacon + # has joined me at outer. + self.assert_location_state('outer') + self.assert_mobile_tracker_state('outer', 'unknown') + + def test_event_beacon_entry_zone_loading_dash(self): + """Test the event for beacon zone landing.""" + # Make sure the leading - is ignored + # Ownracks uses this to switch on hold + + message = build_message( + {'desc': "-inner"}, + REGION_BEACON_ENTER_MESSAGE) + self.send_message(EVENT_TOPIC, message) + self.assert_location_state('inner') + + # ------------------------------------------------------------------------ + # Mobile Beacon based event entry / exit testing + def test_mobile_enter_move_beacon(self): """Test the movement of a beacon.""" - # Enter mobile beacon, should set location - message = REGION_ENTER_MESSAGE.copy() - message['desc'] = IBEACON_DEVICE - self.send_message(EVENT_TOPIC, message) + # I am in the outer zone. + self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE) - self.assert_tracker_latitude(2.0) - self.assert_tracker_state('outer') + # I see the 'keys' beacon. I set the location of the + # beacon_keys tracker to my current device location. + self.send_message(EVENT_TOPIC, MOBILE_BEACON_ENTER_EVENT_MESSAGE) - # Move should move beacon - message = LOCATION_MESSAGE.copy() - message['lat'] = "3.0" - self.send_message(LOCATION_TOPIC, message) + self.assert_mobile_tracker_latitude(LOCATION_MESSAGE['lat']) + self.assert_mobile_tracker_state('outer') - self.assert_tracker_latitude(3.0) - self.assert_tracker_state(STATE_NOT_HOME) + # Location update to outside of defined zones. + # I am now 'not home' and neither are my keys. + self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE_NOT_HOME) + + self.assert_location_state(STATE_NOT_HOME) + self.assert_mobile_tracker_state(STATE_NOT_HOME) + + not_home_lat = LOCATION_MESSAGE_NOT_HOME['lat'] + self.assert_location_latitude(not_home_lat) + self.assert_mobile_tracker_latitude(not_home_lat) def test_mobile_enter_exit_region_beacon(self): - """Test the enter and the exit of a region beacon.""" - # Start tracking beacon - message = REGION_ENTER_MESSAGE.copy() - message['desc'] = IBEACON_DEVICE - self.send_message(EVENT_TOPIC, message) - self.assert_tracker_latitude(2.0) - self.assert_tracker_state('outer') + """Test the enter and the exit of a mobile beacon.""" + # I am in the outer zone. + self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE) - # Enter location should move beacon - message = REGION_ENTER_MESSAGE.copy() - message['desc'] = "inner_2" - self.send_message(EVENT_TOPIC, message) + # I see a new mobile beacon + self.send_message(EVENT_TOPIC, MOBILE_BEACON_ENTER_EVENT_MESSAGE) + self.assert_mobile_tracker_latitude(OUTER_ZONE['latitude']) + self.assert_mobile_tracker_state('outer') - self.assert_tracker_latitude(2.1) - self.assert_tracker_state('inner_2') + # GPS enter message should move beacon + self.send_message(EVENT_TOPIC, REGION_GPS_ENTER_MESSAGE) - # Exit location should switch to gps - message = REGION_LEAVE_MESSAGE.copy() - message['desc'] = "inner_2" - self.send_message(EVENT_TOPIC, message) - self.assert_tracker_latitude(2.0) + self.assert_mobile_tracker_latitude(INNER_ZONE['latitude']) + self.assert_mobile_tracker_state(REGION_GPS_ENTER_MESSAGE['desc']) + + # Exit inner zone to outer zone should move beacon to + # center of outer zone + self.send_message(EVENT_TOPIC, REGION_GPS_LEAVE_MESSAGE) + self.assert_mobile_tracker_latitude(REGION_GPS_LEAVE_MESSAGE['lat']) + self.assert_mobile_tracker_state('outer') def test_mobile_exit_move_beacon(self): """Test the exit move of a beacon.""" - # Start tracking beacon - message = REGION_ENTER_MESSAGE.copy() - message['desc'] = IBEACON_DEVICE - self.send_message(EVENT_TOPIC, message) - self.assert_tracker_latitude(2.0) - self.assert_tracker_state('outer') + # I am in the outer zone. + self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE) + + # I see a new mobile beacon + self.send_message(EVENT_TOPIC, MOBILE_BEACON_ENTER_EVENT_MESSAGE) + self.assert_mobile_tracker_latitude(OUTER_ZONE['latitude']) + self.assert_mobile_tracker_state('outer') # Exit mobile beacon, should set location - message = REGION_LEAVE_MESSAGE.copy() - message['desc'] = IBEACON_DEVICE - message['lat'] = "3.0" - self.send_message(EVENT_TOPIC, message) + self.send_message(EVENT_TOPIC, MOBILE_BEACON_LEAVE_EVENT_MESSAGE) - self.assert_tracker_latitude(3.0) + self.assert_mobile_tracker_latitude(OUTER_ZONE['latitude']) + self.assert_mobile_tracker_state('outer') # Move after exit should do nothing - message = LOCATION_MESSAGE.copy() - message['lat'] = "4.0" - self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE) - self.assert_tracker_latitude(3.0) + self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE_NOT_HOME) + self.assert_mobile_tracker_latitude(OUTER_ZONE['latitude']) + self.assert_mobile_tracker_state('outer') def test_mobile_multiple_async_enter_exit(self): """Test the multiple entering.""" # Test race condition - enter_message = REGION_ENTER_MESSAGE.copy() - enter_message['desc'] = IBEACON_DEVICE - exit_message = REGION_LEAVE_MESSAGE.copy() - exit_message['desc'] = IBEACON_DEVICE - for _ in range(0, 20): fire_mqtt_message( - self.hass, EVENT_TOPIC, json.dumps(enter_message)) + self.hass, EVENT_TOPIC, + json.dumps(MOBILE_BEACON_ENTER_EVENT_MESSAGE)) fire_mqtt_message( - self.hass, EVENT_TOPIC, json.dumps(exit_message)) + self.hass, EVENT_TOPIC, + json.dumps(MOBILE_BEACON_LEAVE_EVENT_MESSAGE)) fire_mqtt_message( - self.hass, EVENT_TOPIC, json.dumps(enter_message)) + self.hass, EVENT_TOPIC, + json.dumps(MOBILE_BEACON_ENTER_EVENT_MESSAGE)) self.hass.block_till_done() - self.send_message(EVENT_TOPIC, exit_message) - self.assertEqual(self.context.mobile_beacons_active['greg_phone'], []) + self.send_message(EVENT_TOPIC, MOBILE_BEACON_LEAVE_EVENT_MESSAGE) + self.assertEqual(len(self.context.mobile_beacons_active['greg_phone']), + 0) def test_mobile_multiple_enter_exit(self): """Test the multiple entering.""" - # Should only happen if the iphone dies - enter_message = REGION_ENTER_MESSAGE.copy() - enter_message['desc'] = IBEACON_DEVICE - exit_message = REGION_LEAVE_MESSAGE.copy() - exit_message['desc'] = IBEACON_DEVICE + self.send_message(EVENT_TOPIC, MOBILE_BEACON_ENTER_EVENT_MESSAGE) + self.send_message(EVENT_TOPIC, MOBILE_BEACON_ENTER_EVENT_MESSAGE) + self.send_message(EVENT_TOPIC, MOBILE_BEACON_LEAVE_EVENT_MESSAGE) - self.send_message(EVENT_TOPIC, enter_message) - self.send_message(EVENT_TOPIC, enter_message) - self.send_message(EVENT_TOPIC, exit_message) + self.assertEqual(len(self.context.mobile_beacons_active['greg_phone']), + 0) - self.assertEqual(self.context.mobile_beacons_active['greg_phone'], []) + def test_complex_movement(self): + """Test a complex sequence representative of real-world use.""" + # I am in the outer zone. + self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE) + self.assert_location_state('outer') + + # gps to inner location and event, as actually happens with OwnTracks + location_message = build_message( + {'lat': REGION_GPS_ENTER_MESSAGE['lat'], + 'lon': REGION_GPS_ENTER_MESSAGE['lon']}, + LOCATION_MESSAGE) + self.send_message(LOCATION_TOPIC, location_message) + self.send_message(EVENT_TOPIC, REGION_GPS_ENTER_MESSAGE) + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_location_state('inner') + + # region beacon enter inner event and location as actually happens + # with OwnTracks + location_message = build_message( + {'lat': location_message['lat'] + FIVE_M, + 'lon': location_message['lon'] + FIVE_M}, + LOCATION_MESSAGE) + self.send_message(EVENT_TOPIC, REGION_BEACON_ENTER_MESSAGE) + self.send_message(LOCATION_TOPIC, location_message) + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_location_state('inner') + + # see keys mobile beacon and location message as actually happens + location_message = build_message( + {'lat': location_message['lat'] + FIVE_M, + 'lon': location_message['lon'] + FIVE_M}, + LOCATION_MESSAGE) + self.send_message(EVENT_TOPIC, MOBILE_BEACON_ENTER_EVENT_MESSAGE) + self.send_message(LOCATION_TOPIC, location_message) + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_mobile_tracker_latitude(INNER_ZONE['latitude']) + self.assert_location_state('inner') + self.assert_mobile_tracker_state('inner') + + # Slightly odd, I leave the location by gps before I lose + # sight of the region beacon. This is also a little odd in + # that my GPS coords are now in the 'outer' zone but I did not + # "enter" that zone when I started up so my location is not + # the center of OUTER_ZONE, but rather just my GPS location. + + # gps out of inner event and location + location_message = build_message( + {'lat': REGION_GPS_LEAVE_MESSAGE['lat'], + 'lon': REGION_GPS_LEAVE_MESSAGE['lon']}, + LOCATION_MESSAGE) + self.send_message(EVENT_TOPIC, REGION_GPS_LEAVE_MESSAGE) + self.send_message(LOCATION_TOPIC, location_message) + self.assert_location_latitude(REGION_GPS_LEAVE_MESSAGE['lat']) + self.assert_mobile_tracker_latitude(REGION_GPS_LEAVE_MESSAGE['lat']) + self.assert_location_state('outer') + self.assert_mobile_tracker_state('outer') + + # region beacon leave inner + location_message = build_message( + {'lat': location_message['lat'] - FIVE_M, + 'lon': location_message['lon'] - FIVE_M}, + LOCATION_MESSAGE) + self.send_message(EVENT_TOPIC, REGION_BEACON_LEAVE_MESSAGE) + self.send_message(LOCATION_TOPIC, location_message) + self.assert_location_latitude(location_message['lat']) + self.assert_mobile_tracker_latitude(location_message['lat']) + self.assert_location_state('outer') + self.assert_mobile_tracker_state('outer') + + # lose keys mobile beacon + lost_keys_location_message = build_message( + {'lat': location_message['lat'] - FIVE_M, + 'lon': location_message['lon'] - FIVE_M}, + LOCATION_MESSAGE) + self.send_message(LOCATION_TOPIC, lost_keys_location_message) + self.send_message(EVENT_TOPIC, MOBILE_BEACON_LEAVE_EVENT_MESSAGE) + self.assert_location_latitude(lost_keys_location_message['lat']) + self.assert_mobile_tracker_latitude(lost_keys_location_message['lat']) + self.assert_location_state('outer') + self.assert_mobile_tracker_state('outer') + + # gps leave outer + self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE_NOT_HOME) + self.send_message(EVENT_TOPIC, REGION_GPS_LEAVE_MESSAGE_OUTER) + self.assert_location_latitude(LOCATION_MESSAGE_NOT_HOME['lat']) + self.assert_mobile_tracker_latitude(lost_keys_location_message['lat']) + self.assert_location_state('not_home') + self.assert_mobile_tracker_state('outer') + + # location move not home + location_message = build_message( + {'lat': LOCATION_MESSAGE_NOT_HOME['lat'] - FIVE_M, + 'lon': LOCATION_MESSAGE_NOT_HOME['lon'] - FIVE_M}, + LOCATION_MESSAGE_NOT_HOME) + self.send_message(LOCATION_TOPIC, location_message) + self.assert_location_latitude(location_message['lat']) + self.assert_mobile_tracker_latitude(lost_keys_location_message['lat']) + self.assert_location_state('not_home') + self.assert_mobile_tracker_state('outer') + + def test_complex_movement_sticky_keys_beacon(self): + """Test a complex sequence which was previously broken.""" + # I am not_home + self.send_message(LOCATION_TOPIC, LOCATION_MESSAGE) + self.assert_location_state('outer') + + # gps to inner location and event, as actually happens with OwnTracks + location_message = build_message( + {'lat': REGION_GPS_ENTER_MESSAGE['lat'], + 'lon': REGION_GPS_ENTER_MESSAGE['lon']}, + LOCATION_MESSAGE) + self.send_message(LOCATION_TOPIC, location_message) + self.send_message(EVENT_TOPIC, REGION_GPS_ENTER_MESSAGE) + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_location_state('inner') + + # see keys mobile beacon and location message as actually happens + location_message = build_message( + {'lat': location_message['lat'] + FIVE_M, + 'lon': location_message['lon'] + FIVE_M}, + LOCATION_MESSAGE) + self.send_message(EVENT_TOPIC, MOBILE_BEACON_ENTER_EVENT_MESSAGE) + self.send_message(LOCATION_TOPIC, location_message) + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_mobile_tracker_latitude(INNER_ZONE['latitude']) + self.assert_location_state('inner') + self.assert_mobile_tracker_state('inner') + + # region beacon enter inner event and location as actually happens + # with OwnTracks + location_message = build_message( + {'lat': location_message['lat'] + FIVE_M, + 'lon': location_message['lon'] + FIVE_M}, + LOCATION_MESSAGE) + self.send_message(EVENT_TOPIC, REGION_BEACON_ENTER_MESSAGE) + self.send_message(LOCATION_TOPIC, location_message) + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_location_state('inner') + + # This sequence of moves would cause keys to follow + # greg_phone around even after the OwnTracks sent + # a mobile beacon 'leave' event for the keys. + # leave keys + self.send_message(LOCATION_TOPIC, location_message) + self.send_message(EVENT_TOPIC, MOBILE_BEACON_LEAVE_EVENT_MESSAGE) + self.assert_location_state('inner') + self.assert_mobile_tracker_state('inner') + + # leave inner region beacon + self.send_message(EVENT_TOPIC, REGION_BEACON_LEAVE_MESSAGE) + self.send_message(LOCATION_TOPIC, location_message) + self.assert_location_state('inner') + self.assert_mobile_tracker_state('inner') + + # enter inner region beacon + self.send_message(EVENT_TOPIC, REGION_BEACON_ENTER_MESSAGE) + self.send_message(LOCATION_TOPIC, location_message) + self.assert_location_latitude(INNER_ZONE['latitude']) + self.assert_location_state('inner') + + # enter keys + self.send_message(EVENT_TOPIC, MOBILE_BEACON_ENTER_EVENT_MESSAGE) + self.send_message(LOCATION_TOPIC, location_message) + self.assert_location_state('inner') + self.assert_mobile_tracker_state('inner') + + # leave keys + self.send_message(LOCATION_TOPIC, location_message) + self.send_message(EVENT_TOPIC, MOBILE_BEACON_LEAVE_EVENT_MESSAGE) + self.assert_location_state('inner') + self.assert_mobile_tracker_state('inner') + + # leave inner region beacon + self.send_message(EVENT_TOPIC, REGION_BEACON_LEAVE_MESSAGE) + self.send_message(LOCATION_TOPIC, location_message) + self.assert_location_state('inner') + self.assert_mobile_tracker_state('inner') + + # GPS leave inner region, I'm in the 'outer' region now + # but on GPS coords + leave_location_message = build_message( + {'lat': REGION_GPS_LEAVE_MESSAGE['lat'], + 'lon': REGION_GPS_LEAVE_MESSAGE['lon']}, + LOCATION_MESSAGE) + self.send_message(EVENT_TOPIC, REGION_GPS_LEAVE_MESSAGE) + self.send_message(LOCATION_TOPIC, leave_location_message) + self.assert_location_state('outer') + self.assert_mobile_tracker_state('inner') + self.assert_location_latitude(REGION_GPS_LEAVE_MESSAGE['lat']) + self.assert_mobile_tracker_latitude(INNER_ZONE['latitude']) def test_waypoint_import_simple(self): """Test a simple import of list of waypoints.""" @@ -698,6 +1118,51 @@ class TestDeviceTrackerOwnTracks(BaseMQTT): self.assertTrue(wayp == new_wayp) +def generate_ciphers(secret): + """Generate test ciphers for the DEFAULT_LOCATION_MESSAGE.""" + # libnacl ciphertext generation will fail if the module + # cannot be imported. However, the test for decryption + # also relies on this library and won't be run without it. + import json + import pickle + import base64 + + try: + from libnacl import crypto_secretbox_KEYBYTES as KEYLEN + from libnacl.secret import SecretBox + key = secret.encode("utf-8")[:KEYLEN].ljust(KEYLEN, b'\0') + ctxt = base64.b64encode(SecretBox(key).encrypt( + json.dumps(DEFAULT_LOCATION_MESSAGE).encode("utf-8")) + ).decode("utf-8") + except (ImportError, OSError): + ctxt = '' + + mctxt = base64.b64encode( + pickle.dumps( + (secret.encode("utf-8"), + json.dumps(DEFAULT_LOCATION_MESSAGE).encode("utf-8")) + ) + ).decode("utf-8") + return (ctxt, mctxt) + + +TEST_SECRET_KEY = 's3cretkey' + +CIPHERTEXT, MOCK_CIPHERTEXT = generate_ciphers(TEST_SECRET_KEY) + +ENCRYPTED_LOCATION_MESSAGE = { + # Encrypted version of LOCATION_MESSAGE using libsodium and TEST_SECRET_KEY + '_type': 'encrypted', + 'data': CIPHERTEXT +} + +MOCK_ENCRYPTED_LOCATION_MESSAGE = { + # Mock-encrypted version of LOCATION_MESSAGE using pickle + '_type': 'encrypted', + 'data': MOCK_CIPHERTEXT +} + + def mock_cipher(): """Return a dummy pickle-based cipher.""" def mock_decrypt(ciphertext, key): @@ -748,7 +1213,7 @@ class TestDeviceTrackerOwnTrackConfigs(BaseMQTT): CONF_SECRET: TEST_SECRET_KEY, }}) self.send_message(LOCATION_TOPIC, MOCK_ENCRYPTED_LOCATION_MESSAGE) - self.assert_location_latitude(2.0) + self.assert_location_latitude(LOCATION_MESSAGE['lat']) @patch('homeassistant.components.device_tracker.owntracks.get_cipher', mock_cipher) @@ -762,7 +1227,7 @@ class TestDeviceTrackerOwnTrackConfigs(BaseMQTT): LOCATION_TOPIC: TEST_SECRET_KEY, }}}) self.send_message(LOCATION_TOPIC, MOCK_ENCRYPTED_LOCATION_MESSAGE) - self.assert_location_latitude(2.0) + self.assert_location_latitude(LOCATION_MESSAGE['lat']) @patch('homeassistant.components.device_tracker.owntracks.get_cipher', mock_cipher) @@ -834,4 +1299,4 @@ class TestDeviceTrackerOwnTrackConfigs(BaseMQTT): }}) self.send_message(LOCATION_TOPIC, ENCRYPTED_LOCATION_MESSAGE) - self.assert_location_latitude(2.0) + self.assert_location_latitude(LOCATION_MESSAGE['lat'])