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

Switch locative to use the webhook component (#20043)

Browse Source 
* Switch locative to use the webhook component

* Lint

* Remove dead test code

* Use voluptuous to validate the webhook schema

* Validate test mode schema as well

* Lint

* Remove allow_extra

* Return web.Response correctly

* #20043: Remove superfluous dict in WEBHOOK_SCHEMA validation
This commit is contained in:
Paulus Schoutsen 2019-01-16 11:08:42 -08:00 committed by GitHub
commit 1d86905d5b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 197 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -0,0 +1,18 @@
{
"config": {
"title": "Locative Webhook",
"step": {
"user": {
"title": "Set up the Locative Webhook",
"description": "Are you sure you want to set up the Locative 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 Geofency."
},
"create_entry": {
"default": "To send locations to Home Assistant, you will need to setup the webhook feature in the Locative app.\n\nFill in the following info:\n\n- URL: `{webhook_url}`\n- Method: POST\n\nSee [the documentation]({docs_url}) for further details."
}
}
}

187
homeassistant/components/locative/__init__.py
View File

@ -5,79 +5,98 @@ For more details about this component, please refer to the documentation at
https://home-assistant.io/components/locative/
"""
import logging
from typing import Dict
import voluptuous as vol
from aiohttp import web
import homeassistant.helpers.config_validation as cv
from homeassistant.components.device_tracker import \
DOMAIN as DEVICE_TRACKER_DOMAIN
from homeassistant.components.http import HomeAssistantView
from homeassistant.const import HTTP_UNPROCESSABLE_ENTITY, ATTR_LATITUDE, \
ATTR_LONGITUDE, STATE_NOT_HOME
ATTR_LONGITUDE, STATE_NOT_HOME, CONF_WEBHOOK_ID, ATTR_ID, HTTP_OK
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 = 'locative'
DEPENDENCIES = ['http']
URL = '/api/locative'
DEPENDENCIES = ['webhook']
TRACKER_UPDATE = '{}_tracker_update'.format(DOMAIN)
ATTR_DEVICE_ID = 'device'
ATTR_TRIGGER = 'trigger'
def _id(value: str) -> str:
"""Coerce id by removing '-'."""
return value.replace('-', '')
def _validate_test_mode(obj: Dict) -> Dict:
"""Validate that id is provided outside of test mode."""
if ATTR_ID not in obj and obj[ATTR_TRIGGER] != 'test':
raise vol.Invalid('Location id not specified')
return obj
WEBHOOK_SCHEMA = vol.All(
vol.Schema({
vol.Required(ATTR_LATITUDE): cv.latitude,
vol.Required(ATTR_LONGITUDE): cv.longitude,
vol.Required(ATTR_DEVICE_ID): cv.string,
vol.Required(ATTR_TRIGGER): cv.string,
vol.Optional(ATTR_ID): vol.All(cv.string, _id)
}),
_validate_test_mode
)
async def async_setup(hass, hass_config):
"""Set up the Locative component."""
hass.http.register_view(LocativeView)
hass.async_create_task(
async_load_platform(hass, 'device_tracker', DOMAIN, {}, hass_config)
)
return True
class LocativeView(HomeAssistantView):
"""View to handle Locative requests."""
async def handle_webhook(hass, webhook_id, request):
"""Handle incoming webhook from Locative."""
try:
data = WEBHOOK_SCHEMA(dict(await request.post()))
except vol.MultipleInvalid as error:
return web.Response(
body=error.error_message,
status=HTTP_UNPROCESSABLE_ENTITY
)
url = URL
name = 'api:locative'
device = data[ATTR_DEVICE_ID]
location_name = data.get(ATTR_ID, data[ATTR_TRIGGER]).lower()
direction = data[ATTR_TRIGGER]
gps_location = (data[ATTR_LATITUDE], data[ATTR_LONGITUDE])
def __init__(self):
"""Initialize Locative URL endpoints."""
if direction == 'enter':
async_dispatcher_send(
hass,
TRACKER_UPDATE,
device,
gps_location,
location_name
)
return web.Response(
body='Setting location to {}'.format(location_name),
status=HTTP_OK
)
async def get(self, request):
"""Locative message received as GET."""
return await self._handle(request.app['hass'], request.query)
if direction == 'exit':
current_state = hass.states.get(
'{}.{}'.format(DEVICE_TRACKER_DOMAIN, device))
async def post(self, request):
"""Locative message received."""
data = await request.post()
return await self._handle(request.app['hass'], data)
async def _handle(self, hass, data):
"""Handle locative request."""
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)
if 'trigger' not in data:
_LOGGER.error('Trigger is not specified.')
return ('Trigger is not specified.',
HTTP_UNPROCESSABLE_ENTITY)
if 'id' not in data and data['trigger'] != 'test':
_LOGGER.error('Location id not specified.')
return ('Location id not specified.',
HTTP_UNPROCESSABLE_ENTITY)
device = data['device'].replace('-', '')
location_name = data.get('id', data['trigger']).lower()
direction = data['trigger']
gps_location = (data[ATTR_LATITUDE], data[ATTR_LONGITUDE])
if direction == 'enter':
if current_state is None or current_state.state == location_name:
location_name = STATE_NOT_HOME
async_dispatcher_send(
hass,
TRACKER_UPDATE,
@ -85,36 +104,54 @@ class LocativeView(HomeAssistantView):
gps_location,
location_name
)
return 'Setting location to {}'.format(location_name)
return web.Response(
body='Setting location to not home',
status=HTTP_OK
)
if direction == 'exit':
current_state = hass.states.get(
'{}.{}'.format(DEVICE_TRACKER_DOMAIN, device))
# 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.
return web.Response(
body='Ignoring exit from {} (already in {})'.format(
location_name, current_state
),
status=HTTP_OK
)
if current_state is None or current_state.state == location_name:
location_name = STATE_NOT_HOME
async_dispatcher_send(
hass,
TRACKER_UPDATE,
device,
gps_location,
location_name
)
return 'Setting location to not home'
if direction == 'test':
# In the app, a test message can be sent. Just return something to
# the user to let them know that it works.
return web.Response(
body='Received test message.',
status=HTTP_OK
)
# 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.
return 'Ignoring exit from {} (already in {})'.format(
location_name, current_state)
_LOGGER.error('Received unidentified message from Locative: %s',
direction)
return web.Response(
body='Received unidentified message: {}'.format(direction),
status=HTTP_UNPROCESSABLE_ENTITY
)
if direction == 'test':
# In the app, a test message can be sent. Just return something to
# the user to let them know that it works.
return 'Received test message.'
_LOGGER.error('Received unidentified message from Locative: %s',
direction)
return ('Received unidentified message: {}'.format(direction),
HTTP_UNPROCESSABLE_ENTITY)
async def async_setup_entry(hass, entry):
"""Configure based on config entry."""
hass.components.webhook.async_register(
DOMAIN, 'Locative', 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,
'Locative Webhook',
{
'docs_url': 'https://www.home-assistant.io/components/locative/'
}
)

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

@ -0,0 +1,18 @@
{
"config": {
"title": "Locative Webhook",
"step": {
"user": {
"title": "Set up the Locative Webhook",
"description": "Are you sure you want to set up the Locative 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 Geofency."
},
"create_entry": {
"default": "To send locations to Home Assistant, you will need to setup the webhook feature in the Locative app.\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

@ -149,6 +149,7 @@ FLOWS = [
'ifttt',
'ios',
'lifx',
'locative',
'luftdaten',
'mailgun',
'mqtt',

79
tests/components/locative/test_init.py
View File

@ -1,23 +1,16 @@
"""The tests the for Locative 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.device_tracker import \
DOMAIN as DEVICE_TRACKER_DOMAIN
from homeassistant.components.locative import URL, DOMAIN
from homeassistant.components.locative import DOMAIN
from homeassistant.const import HTTP_OK, HTTP_UNPROCESSABLE_ENTITY
from homeassistant.setup import async_setup_component
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."""
@ -36,8 +29,26 @@ def locative_client(loop, hass, hass_client):
yield loop.run_until_complete(hass_client())
async def test_missing_data(locative_client):
@pytest.fixture
async def webhook_id(hass, locative_client):
"""Initialize the Geofency component and get the webhook_id."""
hass.config.api = Mock(base_url='http://example.com')
result = await hass.config_entries.flow.async_init('locative', 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(locative_client, webhook_id):
"""Test missing data."""
url = '/api/webhook/{}'.format(webhook_id)
data = {
'latitude': 1.0,
'longitude': 1.1,
@ -47,55 +58,57 @@ async def test_missing_data(locative_client):
}
# No data
req = await locative_client.get(_url({}))
req = await locative_client.post(url)
assert req.status == HTTP_UNPROCESSABLE_ENTITY
# No latitude
copy = data.copy()
del copy['latitude']
req = await locative_client.get(_url(copy))
req = await locative_client.post(url, data=copy)
assert req.status == HTTP_UNPROCESSABLE_ENTITY
# No device
copy = data.copy()
del copy['device']
req = await locative_client.get(_url(copy))
req = await locative_client.post(url, data=copy)
assert req.status == HTTP_UNPROCESSABLE_ENTITY
# No location
copy = data.copy()
del copy['id']
req = await locative_client.get(_url(copy))
req = await locative_client.post(url, data=copy)
assert req.status == HTTP_UNPROCESSABLE_ENTITY
# No trigger
copy = data.copy()
del copy['trigger']
req = await locative_client.get(_url(copy))
req = await locative_client.post(url, data=copy)
assert req.status == HTTP_UNPROCESSABLE_ENTITY
# Test message
copy = data.copy()
copy['trigger'] = 'test'
req = await locative_client.get(_url(copy))
req = await locative_client.post(url, data=copy)
assert req.status == HTTP_OK
# Test message, no location
copy = data.copy()
copy['trigger'] = 'test'
del copy['id']
req = await locative_client.get(_url(copy))
req = await locative_client.post(url, data=copy)
assert req.status == HTTP_OK
# Unknown trigger
copy = data.copy()
copy['trigger'] = 'foobar'
req = await locative_client.get(_url(copy))
req = await locative_client.post(url, data=copy)
assert req.status == HTTP_UNPROCESSABLE_ENTITY
async def test_enter_and_exit(hass, locative_client):
async def test_enter_and_exit(hass, locative_client, webhook_id):
"""Test when there is a known zone."""
url = '/api/webhook/{}'.format(webhook_id)
data = {
'latitude': 40.7855,
'longitude': -111.7367,
@ -105,7 +118,7 @@ async def test_enter_and_exit(hass, locative_client):
}
# Enter the Home
req = await locative_client.get(_url(data))
req = await locative_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,
@ -116,7 +129,7 @@ async def test_enter_and_exit(hass, locative_client):
data['trigger'] = 'exit'
# Exit Home
req = await locative_client.get(_url(data))
req = await locative_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,
@ -127,7 +140,7 @@ async def test_enter_and_exit(hass, locative_client):
data['trigger'] = 'enter'
# Enter Home again
req = await locative_client.get(_url(data))
req = await locative_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,
@ -137,7 +150,7 @@ async def test_enter_and_exit(hass, locative_client):
data['trigger'] = 'exit'
# Exit Home
req = await locative_client.get(_url(data))
req = await locative_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,
@ -148,7 +161,7 @@ async def test_enter_and_exit(hass, locative_client):
data['trigger'] = 'enter'
# Enter Work
req = await locative_client.get(_url(data))
req = await locative_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,
@ -156,8 +169,10 @@ async def test_enter_and_exit(hass, locative_client):
assert 'work' == state_name
async def test_exit_after_enter(hass, locative_client):
async def test_exit_after_enter(hass, locative_client, webhook_id):
"""Test when an exit message comes after an enter message."""
url = '/api/webhook/{}'.format(webhook_id)
data = {
'latitude': 40.7855,
'longitude': -111.7367,
@ -167,7 +182,7 @@ async def test_exit_after_enter(hass, locative_client):
}
# Enter Home
req = await locative_client.get(_url(data))
req = await locative_client.post(url, data=data)
await hass.async_block_till_done()
assert req.status == HTTP_OK
@ -178,7 +193,7 @@ async def test_exit_after_enter(hass, locative_client):
data['id'] = 'Work'
# Enter Work
req = await locative_client.get(_url(data))
req = await locative_client.post(url, data=data)
await hass.async_block_till_done()
assert req.status == HTTP_OK
@ -190,7 +205,7 @@ async def test_exit_after_enter(hass, locative_client):
data['trigger'] = 'exit'
# Exit Home
req = await locative_client.get(_url(data))
req = await locative_client.post(url, data=data)
await hass.async_block_till_done()
assert req.status == HTTP_OK
@ -199,8 +214,10 @@ async def test_exit_after_enter(hass, locative_client):
assert state.state == 'work'
async def test_exit_first(hass, locative_client):
async def test_exit_first(hass, locative_client, webhook_id):
"""Test when an exit message is sent first on a new device."""
url = '/api/webhook/{}'.format(webhook_id)
data = {
'latitude': 40.7855,
'longitude': -111.7367,
@ -210,7 +227,7 @@ async def test_exit_first(hass, locative_client):
}
# Exit Home
req = await locative_client.get(_url(data))
req = await locative_client.post(url, data=data)
await hass.async_block_till_done()
assert req.status == HTTP_OK