Split locative to a separate component (#19964)

* Split locative to a separate component

* Switch tests to use constants for http codes

* Fix tests
This commit is contained in:
Rohan Kapoor 2019-01-11 15:14:11 -08:00 committed by Paulus Schoutsen
parent 8755389c49
commit d820efc4e3
4 changed files with 202 additions and 153 deletions

View File

@ -4,106 +4,25 @@ Support for the Locative platform.
For more details about this platform, please refer to the documentation at For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/device_tracker.locative/ https://home-assistant.io/components/device_tracker.locative/
""" """
from functools import partial
import logging import logging
from homeassistant.const import ( from homeassistant.components.locative import TRACKER_UPDATE
ATTR_LATITUDE, ATTR_LONGITUDE, STATE_NOT_HOME, HTTP_UNPROCESSABLE_ENTITY) from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.components.http import HomeAssistantView
# pylint: disable=unused-import
from homeassistant.components.device_tracker import ( # NOQA
DOMAIN, PLATFORM_SCHEMA)
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
DEPENDENCIES = ['http'] DEPENDENCIES = ['locative']
URL = '/api/locative'
def setup_scanner(hass, config, see, discovery_info=None): async def async_setup_scanner(hass, config, async_see, discovery_info=None):
"""Set up an endpoint for the Locative application.""" """Set up an endpoint for the Locative device tracker."""
hass.http.register_view(LocativeView(see)) async def _set_location(device, gps_location, location_name):
"""Fire HA event to set location."""
await async_see(
dev_id=device,
gps=gps_location,
location_name=location_name
)
async_dispatcher_connect(hass, TRACKER_UPDATE, _set_location)
return True return True
class LocativeView(HomeAssistantView):
"""View to handle Locative requests."""
url = URL
name = 'api:locative'
def __init__(self, see):
"""Initialize Locative URL endpoints."""
self.see = see
async def get(self, request):
"""Locative message received as GET."""
res = await self._handle(request.app['hass'], request.query)
return res
async def post(self, request):
"""Locative message received."""
data = await request.post()
res = await self._handle(request.app['hass'], data)
return res
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':
await hass.async_add_job(
partial(self.see, dev_id=device, location_name=location_name,
gps=gps_location))
return 'Setting location to {}'.format(location_name)
if direction == 'exit':
current_state = hass.states.get(
'{}.{}'.format(DOMAIN, device))
if current_state is None or current_state.state == location_name:
location_name = STATE_NOT_HOME
await hass.async_add_job(
partial(self.see, dev_id=device,
location_name=location_name, gps=gps_location))
return 'Setting location to not home'
# 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)
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)

View File

@ -0,0 +1,120 @@
"""
Support for Locative.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/locative/
"""
import logging
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
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'
TRACKER_UPDATE = '{}_tracker_update'.format(DOMAIN)
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."""
url = URL
name = 'api:locative'
def __init__(self):
"""Initialize Locative URL endpoints."""
async def get(self, request):
"""Locative message received as GET."""
return await self._handle(request.app['hass'], request.query)
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':
async_dispatcher_send(
hass,
TRACKER_UPDATE,
device,
gps_location,
location_name
)
return 'Setting location to {}'.format(location_name)
if direction == 'exit':
current_state = hass.states.get(
'{}.{}'.format(DEVICE_TRACKER_DOMAIN, device))
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'
# 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)
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)

View File

@ -0,0 +1 @@
"""Tests for the Locative component."""

View File

@ -1,13 +1,13 @@
"""The tests the for Locative device tracker platform.""" """The tests the for Locative device tracker platform."""
import asyncio
from unittest.mock import patch from unittest.mock import patch
import pytest import pytest
from homeassistant.components.device_tracker import \
DOMAIN as DEVICE_TRACKER_DOMAIN
from homeassistant.components.locative import URL, DOMAIN
from homeassistant.const import HTTP_OK, HTTP_UNPROCESSABLE_ENTITY
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
import homeassistant.components.device_tracker as device_tracker
from homeassistant.const import CONF_PLATFORM
from homeassistant.components.device_tracker.locative import URL
def _url(data=None): def _url(data=None):
@ -18,22 +18,25 @@ def _url(data=None):
return "{}?{}".format(URL, data) return "{}?{}".format(URL, data)
@pytest.fixture(autouse=True)
def mock_dev_track(mock_device_tracker_conf):
"""Mock device tracker config loading."""
pass
@pytest.fixture @pytest.fixture
def locative_client(loop, hass, hass_client): def locative_client(loop, hass, hass_client):
"""Locative mock client.""" """Locative mock client."""
assert loop.run_until_complete(async_setup_component( assert loop.run_until_complete(async_setup_component(
hass, device_tracker.DOMAIN, { hass, DOMAIN, {
device_tracker.DOMAIN: { DOMAIN: {}
CONF_PLATFORM: 'locative'
}
})) }))
with patch('homeassistant.components.device_tracker.update_config'): with patch('homeassistant.components.device_tracker.update_config'):
yield loop.run_until_complete(hass_client()) yield loop.run_until_complete(hass_client())
@asyncio.coroutine async def test_missing_data(locative_client):
def test_missing_data(locative_client):
"""Test missing data.""" """Test missing data."""
data = { data = {
'latitude': 1.0, 'latitude': 1.0,
@ -44,55 +47,54 @@ def test_missing_data(locative_client):
} }
# No data # No data
req = yield from locative_client.get(_url({})) req = await locative_client.get(_url({}))
assert req.status == 422 assert req.status == HTTP_UNPROCESSABLE_ENTITY
# No latitude # No latitude
copy = data.copy() copy = data.copy()
del copy['latitude'] del copy['latitude']
req = yield from locative_client.get(_url(copy)) req = await locative_client.get(_url(copy))
assert req.status == 422 assert req.status == HTTP_UNPROCESSABLE_ENTITY
# No device # No device
copy = data.copy() copy = data.copy()
del copy['device'] del copy['device']
req = yield from locative_client.get(_url(copy)) req = await locative_client.get(_url(copy))
assert req.status == 422 assert req.status == HTTP_UNPROCESSABLE_ENTITY
# No location # No location
copy = data.copy() copy = data.copy()
del copy['id'] del copy['id']
req = yield from locative_client.get(_url(copy)) req = await locative_client.get(_url(copy))
assert req.status == 422 assert req.status == HTTP_UNPROCESSABLE_ENTITY
# No trigger # No trigger
copy = data.copy() copy = data.copy()
del copy['trigger'] del copy['trigger']
req = yield from locative_client.get(_url(copy)) req = await locative_client.get(_url(copy))
assert req.status == 422 assert req.status == HTTP_UNPROCESSABLE_ENTITY
# Test message # Test message
copy = data.copy() copy = data.copy()
copy['trigger'] = 'test' copy['trigger'] = 'test'
req = yield from locative_client.get(_url(copy)) req = await locative_client.get(_url(copy))
assert req.status == 200 assert req.status == HTTP_OK
# Test message, no location # Test message, no location
copy = data.copy() copy = data.copy()
copy['trigger'] = 'test' copy['trigger'] = 'test'
del copy['id'] del copy['id']
req = yield from locative_client.get(_url(copy)) req = await locative_client.get(_url(copy))
assert req.status == 200 assert req.status == HTTP_OK
# Unknown trigger # Unknown trigger
copy = data.copy() copy = data.copy()
copy['trigger'] = 'foobar' copy['trigger'] = 'foobar'
req = yield from locative_client.get(_url(copy)) req = await locative_client.get(_url(copy))
assert req.status == 422 assert req.status == HTTP_UNPROCESSABLE_ENTITY
@asyncio.coroutine async def test_enter_and_exit(hass, locative_client):
def test_enter_and_exit(hass, locative_client):
"""Test when there is a known zone.""" """Test when there is a known zone."""
data = { data = {
'latitude': 40.7855, 'latitude': 40.7855,
@ -103,9 +105,10 @@ def test_enter_and_exit(hass, locative_client):
} }
# Enter the Home # Enter the Home
req = yield from locative_client.get(_url(data)) req = await locative_client.get(_url(data))
assert req.status == 200 await hass.async_block_till_done()
state_name = hass.states.get('{}.{}'.format('device_tracker', assert req.status == HTTP_OK
state_name = hass.states.get('{}.{}'.format(DEVICE_TRACKER_DOMAIN,
data['device'])).state data['device'])).state
assert 'home' == state_name assert 'home' == state_name
@ -113,9 +116,10 @@ def test_enter_and_exit(hass, locative_client):
data['trigger'] = 'exit' data['trigger'] = 'exit'
# Exit Home # Exit Home
req = yield from locative_client.get(_url(data)) req = await locative_client.get(_url(data))
assert req.status == 200 await hass.async_block_till_done()
state_name = hass.states.get('{}.{}'.format('device_tracker', assert req.status == HTTP_OK
state_name = hass.states.get('{}.{}'.format(DEVICE_TRACKER_DOMAIN,
data['device'])).state data['device'])).state
assert 'not_home' == state_name assert 'not_home' == state_name
@ -123,18 +127,20 @@ def test_enter_and_exit(hass, locative_client):
data['trigger'] = 'enter' data['trigger'] = 'enter'
# Enter Home again # Enter Home again
req = yield from locative_client.get(_url(data)) req = await locative_client.get(_url(data))
assert req.status == 200 await hass.async_block_till_done()
state_name = hass.states.get('{}.{}'.format('device_tracker', assert req.status == HTTP_OK
state_name = hass.states.get('{}.{}'.format(DEVICE_TRACKER_DOMAIN,
data['device'])).state data['device'])).state
assert 'home' == state_name assert 'home' == state_name
data['trigger'] = 'exit' data['trigger'] = 'exit'
# Exit Home # Exit Home
req = yield from locative_client.get(_url(data)) req = await locative_client.get(_url(data))
assert req.status == 200 await hass.async_block_till_done()
state_name = hass.states.get('{}.{}'.format('device_tracker', assert req.status == HTTP_OK
state_name = hass.states.get('{}.{}'.format(DEVICE_TRACKER_DOMAIN,
data['device'])).state data['device'])).state
assert 'not_home' == state_name assert 'not_home' == state_name
@ -142,15 +148,15 @@ def test_enter_and_exit(hass, locative_client):
data['trigger'] = 'enter' data['trigger'] = 'enter'
# Enter Work # Enter Work
req = yield from locative_client.get(_url(data)) req = await locative_client.get(_url(data))
assert req.status == 200 await hass.async_block_till_done()
state_name = hass.states.get('{}.{}'.format('device_tracker', assert req.status == HTTP_OK
state_name = hass.states.get('{}.{}'.format(DEVICE_TRACKER_DOMAIN,
data['device'])).state data['device'])).state
assert 'work' == state_name assert 'work' == state_name
@asyncio.coroutine async def test_exit_after_enter(hass, locative_client):
def test_exit_after_enter(hass, locative_client):
"""Test when an exit message comes after an enter message.""" """Test when an exit message comes after an enter message."""
data = { data = {
'latitude': 40.7855, 'latitude': 40.7855,
@ -161,20 +167,22 @@ def test_exit_after_enter(hass, locative_client):
} }
# Enter Home # Enter Home
req = yield from locative_client.get(_url(data)) req = await locative_client.get(_url(data))
assert req.status == 200 await hass.async_block_till_done()
assert req.status == HTTP_OK
state = hass.states.get('{}.{}'.format('device_tracker', state = hass.states.get('{}.{}'.format(DEVICE_TRACKER_DOMAIN,
data['device'])) data['device']))
assert state.state == 'home' assert state.state == 'home'
data['id'] = 'Work' data['id'] = 'Work'
# Enter Work # Enter Work
req = yield from locative_client.get(_url(data)) req = await locative_client.get(_url(data))
assert req.status == 200 await hass.async_block_till_done()
assert req.status == HTTP_OK
state = hass.states.get('{}.{}'.format('device_tracker', state = hass.states.get('{}.{}'.format(DEVICE_TRACKER_DOMAIN,
data['device'])) data['device']))
assert state.state == 'work' assert state.state == 'work'
@ -182,16 +190,16 @@ def test_exit_after_enter(hass, locative_client):
data['trigger'] = 'exit' data['trigger'] = 'exit'
# Exit Home # Exit Home
req = yield from locative_client.get(_url(data)) req = await locative_client.get(_url(data))
assert req.status == 200 await hass.async_block_till_done()
assert req.status == HTTP_OK
state = hass.states.get('{}.{}'.format('device_tracker', state = hass.states.get('{}.{}'.format(DEVICE_TRACKER_DOMAIN,
data['device'])) data['device']))
assert state.state == 'work' assert state.state == 'work'
@asyncio.coroutine async def test_exit_first(hass, locative_client):
def test_exit_first(hass, locative_client):
"""Test when an exit message is sent first on a new device.""" """Test when an exit message is sent first on a new device."""
data = { data = {
'latitude': 40.7855, 'latitude': 40.7855,
@ -202,9 +210,10 @@ def test_exit_first(hass, locative_client):
} }
# Exit Home # Exit Home
req = yield from locative_client.get(_url(data)) req = await locative_client.get(_url(data))
assert req.status == 200 await hass.async_block_till_done()
assert req.status == HTTP_OK
state = hass.states.get('{}.{}'.format('device_tracker', state = hass.states.get('{}.{}'.format(DEVICE_TRACKER_DOMAIN,
data['device'])) data['device']))
assert state.state == 'not_home' assert state.state == 'not_home'