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

Split out gpslogger into a separate component and platform (#20044) 

* Split out gpslogger into a separate component and platform

* Lint

* Lint

* Increase test coverage
pull/20080/head
Rohan Kapoor 2019-01-13 16:09:47 -08:00 committed by Martin Hjelmare
parent 2a2318b7f6
commit 7f3871028d
5 changed files with 316 additions and 89 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -525,7 +525,6 @@ omit =
homeassistant/components/device_tracker/fritz.py
homeassistant/components/device_tracker/google_maps.py
homeassistant/components/device_tracker/googlehome.py
homeassistant/components/device_tracker/gpslogger.py
homeassistant/components/device_tracker/hitron_coda.py
homeassistant/components/device_tracker/huawei_router.py
homeassistant/components/device_tracker/icloud.py

100
homeassistant/components/device_tracker/gpslogger.py
View File

@ -5,104 +5,28 @@ For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/device_tracker.gpslogger/
"""
import logging
from hmac import compare_digest
from aiohttp.web import Request, HTTPUnauthorized
import voluptuous as vol
import homeassistant.helpers.config_validation as cv
from homeassistant.const import (
CONF_PASSWORD, HTTP_UNPROCESSABLE_ENTITY
)
from homeassistant.components.http import (
CONF_API_PASSWORD, HomeAssistantView
)
# pylint: disable=unused-import
from homeassistant.components.device_tracker import ( # NOQA
DOMAIN, PLATFORM_SCHEMA
)
from homeassistant.components.gpslogger import TRACKER_UPDATE
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.typing import HomeAssistantType, ConfigType
_LOGGER = logging.getLogger(__name__)
DEPENDENCIES = ['http']
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Optional(CONF_PASSWORD): cv.string,
})
DEPENDENCIES = ['gpslogger']
async def async_setup_scanner(hass: HomeAssistantType, config: ConfigType,
async_see, discovery_info=None):
"""Set up an endpoint for the GPSLogger application."""
hass.http.register_view(GPSLoggerView(async_see, config))
return True
class GPSLoggerView(HomeAssistantView):
"""View to handle GPSLogger requests."""
url = '/api/gpslogger'
name = 'api:gpslogger'
def __init__(self, async_see, config):
"""Initialize GPSLogger url endpoints."""
self.async_see = async_see
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']
hass.async_create_task(self.async_see(
"""Set up an endpoint for the GPSLogger device tracker."""
async def _set_location(device, gps_location, battery, accuracy, attrs):
"""Fire HA event to set location."""
await async_see(
dev_id=device,
gps=gps_location, battery=battery,
gps=gps_location,
battery=battery,
gps_accuracy=accuracy,
attributes=attrs
))
)
return 'Setting location for {}'.format(device)
async_dispatcher_connect(hass, TRACKER_UPDATE, _set_location)
return True

114
homeassistant/components/gpslogger/__init__.py Normal file
View File

@ -0,0 +1,114 @@
"""
Support for GPSLogger.
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
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.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)
TRACKER_UPDATE = '{}_tracker_update'.format(DOMAIN)
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
)
return 'Setting location for {}'.format(device)

1
tests/components/gpslogger/__init__.py Normal file
View File

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

189
tests/components/gpslogger/test_init.py Normal file
View File

@ -0,0 +1,189 @@
"""The tests the for GPSLogger device tracker platform."""
from unittest.mock import patch
import pytest
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.const import HTTP_OK, HTTP_UNPROCESSABLE_ENTITY, \
STATE_HOME, STATE_NOT_HOME, HTTP_UNAUTHORIZED, CONF_PASSWORD
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."""
pass
@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)."""
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'
}
}))
with patch('homeassistant.components.device_tracker.update_config'):
yield loop.run_until_complete(aiohttp_client(hass.http.app))
@pytest.fixture(autouse=True)
def setup_zones(loop, hass):
"""Set up Zone config in HA."""
assert loop.run_until_complete(async_setup_component(
hass, zone.DOMAIN, {
'zone': {
'name': 'Home',
'latitude': HOME_LATITUDE,
'longitude': HOME_LONGITUDE,
'radius': 100,
}}))
async def test_authentication(hass, unauthenticated_gpslogger_client):
"""Test missing data."""
data = {
'latitude': 1.0,
'longitude': 1.1,
'device': '123',
CONF_API_PASSWORD: 'test'
}
# 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,
'device': '123',
}
# No data
req = await authenticated_gpslogger_client.get(_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))
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))
await hass.async_block_till_done()
assert req.status == HTTP_UNPROCESSABLE_ENTITY
async def test_enter_and_exit(hass, authenticated_gpslogger_client):
"""Test when there is a known zone."""
data = {
'latitude': HOME_LATITUDE,
'longitude': HOME_LONGITUDE,
'device': '123',
}
# Enter the Home
req = await authenticated_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_HOME == state_name
# Enter Home again
req = await authenticated_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_HOME == state_name
data['longitude'] = 0
data['latitude'] = 0
# Enter Somewhere else
req = await authenticated_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_enter_with_attrs(hass, authenticated_gpslogger_client):
"""Test when additional attributes are present."""
data = {
'latitude': 1.0,
'longitude': 1.1,
'device': '123',
'accuracy': 10.5,
'battery': 10,
'speed': 100,
'direction': 105.32,
'altitude': 102,
'provider': 'gps',
'activity': 'running'
}
req = await authenticated_gpslogger_client.get(_url(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.0 == state.attributes['battery']
assert 100.0 == state.attributes['speed']
assert 105.32 == state.attributes['direction']
assert 102.0 == state.attributes['altitude']
assert 'gps' == state.attributes['provider']
assert 'running' == state.attributes['activity']