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

Merge pull request #825 from philipbl/locative 

Update Locative component
pull/826/head
Philip Lundrigan 2015-12-31 13:10:11 -07:00
parent da77edf8fc 394c87c40b
commit 326e26fbeb
4 changed files with 287 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -45,7 +45,6 @@ omit =
homeassistant/components/device_tracker/asuswrt.py
homeassistant/components/device_tracker/ddwrt.py
homeassistant/components/device_tracker/fritz.py
homeassistant/components/device_tracker/locative.py
homeassistant/components/device_tracker/icloud.py
homeassistant/components/device_tracker/luci.py
homeassistant/components/device_tracker/netgear.py

105
homeassistant/components/device_tracker/locative.py
View File

@ -6,65 +6,100 @@ Locative platform for the device tracker.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/device_tracker.locative/
"""
import logging
from functools import partial
from homeassistant.const import (
HTTP_UNPROCESSABLE_ENTITY, HTTP_INTERNAL_SERVER_ERROR)
HTTP_UNPROCESSABLE_ENTITY, STATE_NOT_HOME)
from homeassistant.components.device_tracker import DOMAIN
_LOGGER = logging.getLogger(__name__)
DEPENDENCIES = ['http']
_SEE = 0
URL_API_LOCATIVE_ENDPOINT = "/api/locative"
def setup_scanner(hass, config, see):
""" Set up an endpoint for the Locative app. """
# Use a global variable to keep setup_scanner compact when using a callback
global _SEE
_SEE = see
# POST would be semantically better, but that currently does not work
# since Locative sends the data as key1=value1&key2=value2
# in the request body, while Home Assistant expects json there.
hass.http.register_path(
'GET', URL_API_LOCATIVE_ENDPOINT, _handle_get_api_locative)
'GET', URL_API_LOCATIVE_ENDPOINT,
partial(_handle_get_api_locative, hass, see))
return True
def _handle_get_api_locative(handler, path_match, data):
def _handle_get_api_locative(hass, see, handler, path_match, data):
""" Locative message received. """
if not isinstance(data, dict):
handler.write_json_message(
"Error while parsing Locative message.",
HTTP_INTERNAL_SERVER_ERROR)
if not _check_data(handler, data):
return
device = data['device'].replace('-', '')
location_name = data['id'].lower()
direction = data['trigger']
if direction == 'enter':
see(dev_id=device, location_name=location_name)
handler.write_text("Setting location to {}".format(location_name))
elif direction == 'exit':
current_state = hass.states.get(
"{}.{}".format(DOMAIN, device)).state
if current_state == location_name:
see(dev_id=device, location_name=STATE_NOT_HOME)
handler.write_text("Setting location to not home")
else:
# Ignore the message if it is telling us to exit a zone that we
# aren't currently in. This occurs when a zone is entered before
# the previous zone was exited. The enter message will be sent
# first, then the exit message will be sent second.
handler.write_text(
'Ignoring exit from {} (already in {})'.format(
location_name, current_state))
elif direction == 'test':
# In the app, a test message can be sent. Just return something to
# the user to let them know that it works.
handler.write_text("Received test message.")
else:
handler.write_text(
"Received unidentified message: {}".format(direction),
HTTP_UNPROCESSABLE_ENTITY)
_LOGGER.error("Received unidentified message from Locative: %s",
direction)
def _check_data(handler, data):
if 'latitude' not in data or 'longitude' not in data:
handler.write_json_message(
"Location not specified.",
HTTP_UNPROCESSABLE_ENTITY)
return
if 'device' not in data or 'id' not in data:
handler.write_json_message(
"Device id or location id not specified.",
HTTP_UNPROCESSABLE_ENTITY)
return
handler.write_text("Latitude and longitude not specified.",
HTTP_UNPROCESSABLE_ENTITY)
_LOGGER.error("Latitude and longitude not specified.")
return False
try:
gps_coords = (float(data['latitude']), float(data['longitude']))
except ValueError:
# If invalid latitude / longitude format
handler.write_json_message(
"Invalid latitude / longitude format.",
HTTP_UNPROCESSABLE_ENTITY)
return
if 'device' not in data:
handler.write_text("Device id not specified.",
HTTP_UNPROCESSABLE_ENTITY)
_LOGGER.error("Device id not specified.")
return False
# entity id's in Home Assistant must be alphanumerical
device_uuid = data['device']
device_entity_id = device_uuid.replace('-', '')
if 'id' not in data:
handler.write_text("Location id not specified.",
HTTP_UNPROCESSABLE_ENTITY)
_LOGGER.error("Location id not specified.")
return False
_SEE(dev_id=device_entity_id, gps=gps_coords, location_name=data['id'])
if 'trigger' not in data:
handler.write_text("Trigger is not specified.",
HTTP_UNPROCESSABLE_ENTITY)
_LOGGER.error("Trigger is not specified.")
return False
handler.write_json_message("Locative message processed")
return True

13
homeassistant/components/http.py
View File

@ -21,7 +21,7 @@ from urllib.parse import urlparse, parse_qs
import homeassistant.core as ha
from homeassistant.const import (
SERVER_PORT, CONTENT_TYPE_JSON,
SERVER_PORT, CONTENT_TYPE_JSON, CONTENT_TYPE_TEXT_PLAIN,
HTTP_HEADER_HA_AUTH, HTTP_HEADER_CONTENT_TYPE, HTTP_HEADER_ACCEPT_ENCODING,
HTTP_HEADER_CONTENT_ENCODING, HTTP_HEADER_VARY, HTTP_HEADER_CONTENT_LENGTH,
HTTP_HEADER_CACHE_CONTROL, HTTP_HEADER_EXPIRES, HTTP_OK, HTTP_UNAUTHORIZED,
@ -293,6 +293,17 @@ class RequestHandler(SimpleHTTPRequestHandler):
json.dumps(data, indent=4, sort_keys=True,
cls=rem.JSONEncoder).encode("UTF-8"))
def write_text(self, message, status_code=HTTP_OK):
""" Helper method to return a text message to the caller. """
self.send_response(status_code)
self.send_header(HTTP_HEADER_CONTENT_TYPE, CONTENT_TYPE_TEXT_PLAIN)
self.set_session_cookie_header()
self.end_headers()
self.wfile.write(message.encode("UTF-8"))
def write_file(self, path, cache_headers=True):
""" Returns a file to the user. """
try:

205
tests/components/device_tracker/test_locative.py Normal file
View File

@ -0,0 +1,205 @@
"""
tests.components.device_tracker.locative
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Tests the locative device tracker component.
"""
import unittest
from unittest.mock import patch
import requests
from homeassistant import bootstrap, const
import homeassistant.core as ha
import homeassistant.components.device_tracker as device_tracker
import homeassistant.components.http as http
import homeassistant.components.zone as zone
SERVER_PORT = 8126
HTTP_BASE_URL = "http://127.0.0.1:{}".format(SERVER_PORT)
hass = None
def _url(data={}):
""" Helper method to generate urls. """
data = "&".join(["{}={}".format(name, value) for name, value in data.items()])
return "{}{}locative?{}".format(HTTP_BASE_URL, const.URL_API, data)
@patch('homeassistant.components.http.util.get_local_ip',
return_value='127.0.0.1')
def setUpModule(mock_get_local_ip): # pylint: disable=invalid-name
""" Initalizes a Home Assistant server. """
global hass
hass = ha.HomeAssistant()
# Set up server
bootstrap.setup_component(hass, http.DOMAIN, {
http.DOMAIN: {
http.CONF_SERVER_PORT: SERVER_PORT
}
})
# Set up API
bootstrap.setup_component(hass, 'api')
# Set up device tracker
bootstrap.setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
'platform': 'locative'
}
})
hass.start()
def tearDownModule(): # pylint: disable=invalid-name
""" Stops the Home Assistant server. """
hass.stop()
# Stub out update_config or else Travis CI raises an exception
@patch('homeassistant.components.device_tracker.update_config')
class TestLocative(unittest.TestCase):
""" Test Locative """
def test_missing_data(self, update_config):
data = {
'latitude': 1.0,
'longitude': 1.1,
'device': '123',
'id': 'Home',
'trigger': 'enter'
}
# No data
req = requests.get(_url({}))
self.assertEqual(422, req.status_code)
# No latitude
copy = data.copy()
del copy['latitude']
req = requests.get(_url(copy))
self.assertEqual(422, req.status_code)
# No device
copy = data.copy()
del copy['device']
req = requests.get(_url(copy))
self.assertEqual(422, req.status_code)
# No location
copy = data.copy()
del copy['id']
req = requests.get(_url(copy))
self.assertEqual(422, req.status_code)
# No trigger
copy = data.copy()
del copy['trigger']
req = requests.get(_url(copy))
self.assertEqual(422, req.status_code)
# Test message
copy = data.copy()
copy['trigger'] = 'test'
req = requests.get(_url(copy))
self.assertEqual(200, req.status_code)
# Unknown trigger
copy = data.copy()
copy['trigger'] = 'foobar'
req = requests.get(_url(copy))
self.assertEqual(422, req.status_code)
def test_enter_and_exit(self, update_config):
""" Test when there is a known zone """
data = {
'latitude': 40.7855,
'longitude': -111.7367,
'device': '123',
'id': 'Home',
'trigger': 'enter'
}
# Enter the Home
req = requests.get(_url(data))
self.assertEqual(200, req.status_code)
state_name = hass.states.get('{}.{}'.format('device_tracker', data['device'])).state
self.assertEqual(state_name, 'home')
data['id'] = 'HOME'
data['trigger'] = 'exit'
# Exit Home
req = requests.get(_url(data))
self.assertEqual(200, req.status_code)
state_name = hass.states.get('{}.{}'.format('device_tracker', data['device'])).state
self.assertEqual(state_name, 'not_home')
data['id'] = 'hOmE'
data['trigger'] = 'enter'
# Enter Home again
req = requests.get(_url(data))
self.assertEqual(200, req.status_code)
state_name = hass.states.get('{}.{}'.format('device_tracker', data['device'])).state
self.assertEqual(state_name, 'home')
data['trigger'] = 'exit'
# Exit Home
req = requests.get(_url(data))
self.assertEqual(200, req.status_code)
state_name = hass.states.get('{}.{}'.format('device_tracker', data['device'])).state
self.assertEqual(state_name, 'not_home')
data['id'] = 'work'
data['trigger'] = 'enter'
# Enter Work
req = requests.get(_url(data))
self.assertEqual(200, req.status_code)
state_name = hass.states.get('{}.{}'.format('device_tracker', data['device'])).state
self.assertEqual(state_name, 'work')
def test_exit_after_enter(self, update_config):
""" Test when an exit message comes after an enter message """
data = {
'latitude': 40.7855,
'longitude': -111.7367,
'device': '123',
'id': 'Home',
'trigger': 'enter'
}
# Enter Home
req = requests.get(_url(data))
self.assertEqual(200, req.status_code)
state = hass.states.get('{}.{}'.format('device_tracker', data['device']))
self.assertEqual(state.state, 'home')
data['id'] = 'Work'
# Enter Work
req = requests.get(_url(data))
self.assertEqual(200, req.status_code)
state = hass.states.get('{}.{}'.format('device_tracker', data['device']))
self.assertEqual(state.state, 'work')
data['id'] = 'Home'
data['trigger'] = 'exit'
# Exit Home
req = requests.get(_url(data))
self.assertEqual(200, req.status_code)
state = hass.states.get('{}.{}'.format('device_tracker', data['device']))
self.assertEqual(state.state, 'work')