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

Migrate gpslogger to the automatically generated webhook (#20079) 

* Migrate gpslogger to the automatically generated webhook

* Lint

* Lint and return error code
pull/20172/head
Rohan Kapoor 2019-01-16 10:56:25 -08:00 committed by Paulus Schoutsen
parent 075b575bde
commit b5bfc759ec
5 changed files with 165 additions and 145 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
homeassistant/components/gpslogger/.translations/en.json Normal file
View File

@ -0,0 +1,18 @@
{
"config": {
"title": "GPSLogger Webhook",
"step": {
"user": {
"title": "Set up the GPSLogger Webhook",
"description": "Are you sure you want to set up the GPSLogger Webhook?"
}
},
"abort": {
"one_instance_allowed": "Only a single instance is necessary.",
"not_internet_accessible": "Your Home Assistant instance needs to be accessible from the internet to receive messages from GPSLogger."
},
"create_entry": {
"default": "To send events to Home Assistant, you will need to setup the webhook feature in GPSLogger.\n\nFill in the following info:\n\n- URL: `{webhook_url}`\n- Method: POST\n\nSee [the documentation]({docs_url}) for further details."
}
}
}

172
homeassistant/components/gpslogger/__init__.py
View File

@ -5,110 +5,116 @@ For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/gpslogger/
"""
import logging
from hmac import compare_digest
import voluptuous as vol
from aiohttp.web_exceptions import HTTPUnauthorized
from aiohttp.web_request import Request
from aiohttp import web
import homeassistant.helpers.config_validation as cv
from homeassistant.components.http import HomeAssistantView, CONF_API_PASSWORD
from homeassistant.const import CONF_PASSWORD, HTTP_UNPROCESSABLE_ENTITY
from homeassistant.components.device_tracker import ATTR_BATTERY
from homeassistant.components.device_tracker.tile import ATTR_ALTITUDE
from homeassistant.const import HTTP_UNPROCESSABLE_ENTITY, \
HTTP_OK, ATTR_LATITUDE, ATTR_LONGITUDE, CONF_WEBHOOK_ID
from homeassistant.helpers import config_entry_flow
from homeassistant.helpers.discovery import async_load_platform
from homeassistant.helpers.dispatcher import async_dispatcher_send
_LOGGER = logging.getLogger(__name__)
DOMAIN = 'gpslogger'
DEPENDENCIES = ['http']
CONFIG_SCHEMA = vol.Schema({
vol.Optional(DOMAIN): vol.Schema({
vol.Optional(CONF_PASSWORD): cv.string,
}),
}, extra=vol.ALLOW_EXTRA)
URL = '/api/{}'.format(DOMAIN)
DEPENDENCIES = ['webhook']
TRACKER_UPDATE = '{}_tracker_update'.format(DOMAIN)
ATTR_ACCURACY = 'accuracy'
ATTR_ACTIVITY = 'activity'
ATTR_DEVICE = 'device'
ATTR_DIRECTION = 'direction'
ATTR_PROVIDER = 'provider'
ATTR_SPEED = 'speed'
DEFAULT_ACCURACY = 200
DEFAULT_BATTERY = -1
def _id(value: str) -> str:
"""Coerce id by removing '-'."""
return value.replace('-', '')
WEBHOOK_SCHEMA = vol.Schema({
vol.Required(ATTR_LATITUDE): cv.latitude,
vol.Required(ATTR_LONGITUDE): cv.longitude,
vol.Required(ATTR_DEVICE): _id,
vol.Optional(ATTR_ACCURACY, default=DEFAULT_ACCURACY): vol.Coerce(float),
vol.Optional(ATTR_BATTERY, default=DEFAULT_BATTERY): vol.Coerce(float),
vol.Optional(ATTR_SPEED): vol.Coerce(float),
vol.Optional(ATTR_DIRECTION): vol.Coerce(float),
vol.Optional(ATTR_ALTITUDE): vol.Coerce(float),
vol.Optional(ATTR_PROVIDER): cv.string,
vol.Optional(ATTR_ACTIVITY): cv.string
})
async def async_setup(hass, hass_config):
"""Set up the GPSLogger component."""
config = hass_config[DOMAIN]
hass.http.register_view(GPSLoggerView(config))
hass.async_create_task(
async_load_platform(hass, 'device_tracker', DOMAIN, {}, hass_config)
)
return True
class GPSLoggerView(HomeAssistantView):
"""View to handle GPSLogger requests."""
url = URL
name = 'api:gpslogger'
def __init__(self, config):
"""Initialize GPSLogger url endpoints."""
self._password = config.get(CONF_PASSWORD)
# this component does not require external authentication if
# password is set
self.requires_auth = self._password is None
async def get(self, request: Request):
"""Handle for GPSLogger message received as GET."""
hass = request.app['hass']
data = request.query
if self._password is not None:
authenticated = CONF_API_PASSWORD in data and compare_digest(
self._password,
data[CONF_API_PASSWORD]
)
if not authenticated:
raise HTTPUnauthorized()
if 'latitude' not in data or 'longitude' not in data:
return ('Latitude and longitude not specified.',
HTTP_UNPROCESSABLE_ENTITY)
if 'device' not in data:
_LOGGER.error("Device id not specified")
return ('Device id not specified.',
HTTP_UNPROCESSABLE_ENTITY)
device = data['device'].replace('-', '')
gps_location = (data['latitude'], data['longitude'])
accuracy = 200
battery = -1
if 'accuracy' in data:
accuracy = int(float(data['accuracy']))
if 'battery' in data:
battery = float(data['battery'])
attrs = {}
if 'speed' in data:
attrs['speed'] = float(data['speed'])
if 'direction' in data:
attrs['direction'] = float(data['direction'])
if 'altitude' in data:
attrs['altitude'] = float(data['altitude'])
if 'provider' in data:
attrs['provider'] = data['provider']
if 'activity' in data:
attrs['activity'] = data['activity']
async_dispatcher_send(
hass,
TRACKER_UPDATE,
device,
gps_location,
battery,
accuracy,
attrs
async def handle_webhook(hass, webhook_id, request):
"""Handle incoming webhook with GPSLogger request."""
try:
data = WEBHOOK_SCHEMA(dict(await request.post()))
except vol.MultipleInvalid as error:
return web.Response(
body=error.error_message,
status=HTTP_UNPROCESSABLE_ENTITY
)
return 'Setting location for {}'.format(device)
attrs = {
ATTR_SPEED: data.get(ATTR_SPEED),
ATTR_DIRECTION: data.get(ATTR_DIRECTION),
ATTR_ALTITUDE: data.get(ATTR_ALTITUDE),
ATTR_PROVIDER: data.get(ATTR_PROVIDER),
ATTR_ACTIVITY: data.get(ATTR_ACTIVITY)
}
device = data[ATTR_DEVICE]
async_dispatcher_send(
hass,
TRACKER_UPDATE,
device,
(data[ATTR_LATITUDE], data[ATTR_LONGITUDE]),
data[ATTR_BATTERY],
data[ATTR_ACCURACY],
attrs
)
return web.Response(
body='Setting location for {}'.format(device),
status=HTTP_OK
)
async def async_setup_entry(hass, entry):
"""Configure based on config entry."""
hass.components.webhook.async_register(
DOMAIN, 'GPSLogger', entry.data[CONF_WEBHOOK_ID], handle_webhook)
return True
async def async_unload_entry(hass, entry):
"""Unload a config entry."""
hass.components.webhook.async_unregister(entry.data[CONF_WEBHOOK_ID])
return True
config_entry_flow.register_webhook_flow(
DOMAIN,
'GPSLogger Webhook',
{
'docs_url': 'https://www.home-assistant.io/components/gpslogger/'
}
)

18
homeassistant/components/gpslogger/strings.json Normal file
View File

@ -0,0 +1,18 @@
{
"config": {
"title": "GPSLogger Webhook",
"step": {
"user": {
"title": "Set up the GPSLogger Webhook",
"description": "Are you sure you want to set up the GPSLogger Webhook?"
}
},
"abort": {
"one_instance_allowed": "Only a single instance is necessary.",
"not_internet_accessible": "Your Home Assistant instance needs to be accessible from the internet to receive messages from GPSLogger."
},
"create_entry": {
"default": "To send events to Home Assistant, you will need to setup the webhook feature in GPSLogger.\n\nFill in the following info:\n\n- URL: `{webhook_url}`\n- Method: POST\n\nSee [the documentation]({docs_url}) for further details."
}
}
}

1
homeassistant/config_entries.py
View File

@ -142,6 +142,7 @@ FLOWS = [
'esphome',
'emulated_roku',
'geofency',
'gpslogger',
'hangouts',
'homematicip_cloud',
'hue',

101
tests/components/gpslogger/test_init.py
View File

@ -1,29 +1,21 @@
"""The tests the for GPSLogger device tracker platform."""
from unittest.mock import patch
from unittest.mock import patch, Mock
import pytest
from homeassistant import data_entry_flow
from homeassistant.components import zone
from homeassistant.components.device_tracker import \
DOMAIN as DEVICE_TRACKER_DOMAIN
from homeassistant.components.gpslogger import URL, DOMAIN
from homeassistant.components.http import CONF_API_PASSWORD
from homeassistant.components.gpslogger import DOMAIN
from homeassistant.const import HTTP_OK, HTTP_UNPROCESSABLE_ENTITY, \
STATE_HOME, STATE_NOT_HOME, HTTP_UNAUTHORIZED, CONF_PASSWORD
STATE_HOME, STATE_NOT_HOME
from homeassistant.setup import async_setup_component
HOME_LATITUDE = 37.239622
HOME_LONGITUDE = -115.815811
def _url(data=None):
"""Generate URL."""
data = data or {}
data = "&".join(["{}={}".format(name, value) for
name, value in data.items()])
return "{}?{}".format(URL, data)
@pytest.fixture(autouse=True)
def mock_dev_track(mock_device_tracker_conf):
"""Mock device tracker config loading."""
@ -31,28 +23,14 @@ def mock_dev_track(mock_device_tracker_conf):
@pytest.fixture
def authenticated_gpslogger_client(loop, hass, hass_client):
"""Locative mock client (authenticated)."""
assert loop.run_until_complete(async_setup_component(
hass, DOMAIN, {
DOMAIN: {}
}))
with patch('homeassistant.components.device_tracker.update_config'):
yield loop.run_until_complete(hass_client())
@pytest.fixture
def unauthenticated_gpslogger_client(loop, hass, aiohttp_client):
"""Locative mock client (unauthenticated)."""
def gpslogger_client(loop, hass, aiohttp_client):
"""Mock client for GPSLogger (unauthenticated)."""
assert loop.run_until_complete(async_setup_component(
hass, 'persistent_notification', {}))
assert loop.run_until_complete(async_setup_component(
hass, DOMAIN, {
DOMAIN: {
CONF_PASSWORD: 'test'
}
DOMAIN: {}
}))
with patch('homeassistant.components.device_tracker.update_config'):
@ -72,31 +50,26 @@ def setup_zones(loop, hass):
}}))
async def test_authentication(hass, unauthenticated_gpslogger_client):
@pytest.fixture
async def webhook_id(hass, gpslogger_client):
"""Initialize the GPSLogger component and get the webhook_id."""
hass.config.api = Mock(base_url='http://example.com')
result = await hass.config_entries.flow.async_init(DOMAIN, context={
'source': 'user'
})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM, result
result = await hass.config_entries.flow.async_configure(
result['flow_id'], {})
assert result['type'] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
return result['result'].data['webhook_id']
async def test_missing_data(hass, gpslogger_client, webhook_id):
"""Test missing data."""
data = {
'latitude': 1.0,
'longitude': 1.1,
'device': '123',
CONF_API_PASSWORD: 'test'
}
url = '/api/webhook/{}'.format(webhook_id)
# No auth
req = await unauthenticated_gpslogger_client.get(_url({}))
await hass.async_block_till_done()
assert req.status == HTTP_UNAUTHORIZED
# Authenticated
req = await unauthenticated_gpslogger_client.get(_url(data))
await hass.async_block_till_done()
assert req.status == HTTP_OK
state_name = hass.states.get('{}.{}'.format(DEVICE_TRACKER_DOMAIN,
data['device'])).state
assert STATE_NOT_HOME == state_name
async def test_missing_data(hass, authenticated_gpslogger_client):
"""Test missing data."""
data = {
'latitude': 1.0,
'longitude': 1.1,
@ -104,27 +77,29 @@ async def test_missing_data(hass, authenticated_gpslogger_client):
}
# No data
req = await authenticated_gpslogger_client.get(_url({}))
req = await gpslogger_client.post(url)
await hass.async_block_till_done()
assert req.status == HTTP_UNPROCESSABLE_ENTITY
# No latitude
copy = data.copy()
del copy['latitude']
req = await authenticated_gpslogger_client.get(_url(copy))
req = await gpslogger_client.post(url, data=copy)
await hass.async_block_till_done()
assert req.status == HTTP_UNPROCESSABLE_ENTITY
# No device
copy = data.copy()
del copy['device']
req = await authenticated_gpslogger_client.get(_url(copy))
req = await gpslogger_client.post(url, data=copy)
await hass.async_block_till_done()
assert req.status == HTTP_UNPROCESSABLE_ENTITY
async def test_enter_and_exit(hass, authenticated_gpslogger_client):
async def test_enter_and_exit(hass, gpslogger_client, webhook_id):
"""Test when there is a known zone."""
url = '/api/webhook/{}'.format(webhook_id)
data = {
'latitude': HOME_LATITUDE,
'longitude': HOME_LONGITUDE,
@ -132,7 +107,7 @@ async def test_enter_and_exit(hass, authenticated_gpslogger_client):
}
# Enter the Home
req = await authenticated_gpslogger_client.get(_url(data))
req = await gpslogger_client.post(url, data=data)
await hass.async_block_till_done()
assert req.status == HTTP_OK
state_name = hass.states.get('{}.{}'.format(DEVICE_TRACKER_DOMAIN,
@ -140,7 +115,7 @@ async def test_enter_and_exit(hass, authenticated_gpslogger_client):
assert STATE_HOME == state_name
# Enter Home again
req = await authenticated_gpslogger_client.get(_url(data))
req = await gpslogger_client.post(url, data=data)
await hass.async_block_till_done()
assert req.status == HTTP_OK
state_name = hass.states.get('{}.{}'.format(DEVICE_TRACKER_DOMAIN,
@ -151,7 +126,7 @@ async def test_enter_and_exit(hass, authenticated_gpslogger_client):
data['latitude'] = 0
# Enter Somewhere else
req = await authenticated_gpslogger_client.get(_url(data))
req = await gpslogger_client.post(url, data=data)
await hass.async_block_till_done()
assert req.status == HTTP_OK
state_name = hass.states.get('{}.{}'.format(DEVICE_TRACKER_DOMAIN,
@ -159,8 +134,10 @@ async def test_enter_and_exit(hass, authenticated_gpslogger_client):
assert STATE_NOT_HOME == state_name
async def test_enter_with_attrs(hass, authenticated_gpslogger_client):
async def test_enter_with_attrs(hass, gpslogger_client, webhook_id):
"""Test when additional attributes are present."""
url = '/api/webhook/{}'.format(webhook_id)
data = {
'latitude': 1.0,
'longitude': 1.1,
@ -174,13 +151,13 @@ async def test_enter_with_attrs(hass, authenticated_gpslogger_client):
'activity': 'running'
}
req = await authenticated_gpslogger_client.get(_url(data))
req = await gpslogger_client.post(url, data=data)
await hass.async_block_till_done()
assert req.status == HTTP_OK
state = hass.states.get('{}.{}'.format(DEVICE_TRACKER_DOMAIN,
data['device']))
assert STATE_NOT_HOME == state.state
assert 10 == state.attributes['gps_accuracy']
assert 10.5 == state.attributes['gps_accuracy']
assert 10.0 == state.attributes['battery']
assert 100.0 == state.attributes['speed']
assert 105.32 == state.attributes['direction']