jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 08:47:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

OwnTracks Config Entry (#18759)

Browse Source 
* OwnTracks Config Entry

* Fix test

* Fix headers

* Lint

* Username for android only

* Update translations

* Tweak translation

* Create config entry if not there

* Update reqs

* Types

* Lint
This commit is contained in:
Paulus Schoutsen 2018-11-28 22:20:13 +01:00 committed by GitHub
parent e06fa0d2d0
commit 48e28843e6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 554 additions and 355 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
homeassistant/components/device_tracker/__init__.py
View File

@ -181,6 +181,9 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType):
setup = await hass.async_add_job(
platform.setup_scanner, hass, p_config, tracker.see,
disc_info)
elif hasattr(platform, 'async_setup_entry'):
setup = await platform.async_setup_entry(
hass, p_config, tracker.async_see)
else:
raise HomeAssistantError("Invalid device_tracker platform.")
@ -196,6 +199,8 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType):
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Error setting up platform %s", p_type)
hass.data[DOMAIN] = async_setup_platform
setup_tasks = [async_setup_platform(p_type, p_config) for p_type, p_config
in config_per_platform(config, DOMAIN)]
if setup_tasks:
@ -229,6 +234,12 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType):
return True
async def async_setup_entry(hass, entry):
"""Set up an entry."""
await hass.data[DOMAIN](entry.domain, entry)
return True
class DeviceTracker:
"""Representation of a device tracker."""

158
homeassistant/components/device_tracker/owntracks.py
View File

@ -7,55 +7,29 @@ https://home-assistant.io/components/device_tracker.owntracks/
import base64
import json
import logging
from collections import defaultdict
import voluptuous as vol
from homeassistant.components import mqtt
import homeassistant.helpers.config_validation as cv
from homeassistant.components import zone as zone_comp
from homeassistant.components.device_tracker import (
PLATFORM_SCHEMA, ATTR_SOURCE_TYPE, SOURCE_TYPE_BLUETOOTH_LE,
SOURCE_TYPE_GPS
ATTR_SOURCE_TYPE, SOURCE_TYPE_BLUETOOTH_LE, SOURCE_TYPE_GPS
)
from homeassistant.components.owntracks import DOMAIN as OT_DOMAIN
from homeassistant.const import STATE_HOME
from homeassistant.core import callback
from homeassistant.util import slugify, decorator
REQUIREMENTS = ['libnacl==1.6.1']
DEPENDENCIES = ['owntracks']
_LOGGER = logging.getLogger(__name__)
HANDLERS = decorator.Registry()
BEACON_DEV_ID = 'beacon'
CONF_MAX_GPS_ACCURACY = 'max_gps_accuracy'
CONF_SECRET = 'secret'
CONF_WAYPOINT_IMPORT = 'waypoints'
CONF_WAYPOINT_WHITELIST = 'waypoint_whitelist'
CONF_MQTT_TOPIC = 'mqtt_topic'
CONF_REGION_MAPPING = 'region_mapping'
CONF_EVENTS_ONLY = 'events_only'
DEPENDENCIES = ['mqtt']
DEFAULT_OWNTRACKS_TOPIC = 'owntracks/#'
REGION_MAPPING = {}
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Optional(CONF_MAX_GPS_ACCURACY): vol.Coerce(float),
vol.Optional(CONF_WAYPOINT_IMPORT, default=True): cv.boolean,
vol.Optional(CONF_EVENTS_ONLY, default=False): cv.boolean,
vol.Optional(CONF_MQTT_TOPIC, default=DEFAULT_OWNTRACKS_TOPIC):
mqtt.valid_subscribe_topic,
vol.Optional(CONF_WAYPOINT_WHITELIST): vol.All(
cv.ensure_list, [cv.string]),
vol.Optional(CONF_SECRET): vol.Any(
vol.Schema({vol.Optional(cv.string): cv.string}),
cv.string),
vol.Optional(CONF_REGION_MAPPING, default=REGION_MAPPING): dict
})
async def async_setup_entry(hass, entry, async_see):
"""Set up OwnTracks based off an entry."""
hass.data[OT_DOMAIN]['context'].async_see = async_see
hass.helpers.dispatcher.async_dispatcher_connect(
OT_DOMAIN, async_handle_message)
return True
def get_cipher():
@ -72,29 +46,6 @@ def get_cipher():
return (KEYLEN, decrypt)
async def async_setup_scanner(hass, config, async_see, discovery_info=None):
"""Set up an OwnTracks tracker."""
context = context_from_config(async_see, config)
async def async_handle_mqtt_message(topic, payload, qos):
"""Handle incoming OwnTracks message."""
try:
message = json.loads(payload)
except ValueError:
# If invalid JSON
_LOGGER.error("Unable to parse payload as JSON: %s", payload)
return
message['topic'] = topic
await async_handle_message(hass, context, message)
await mqtt.async_subscribe(
hass, context.mqtt_topic, async_handle_mqtt_message, 1)
return True
def _parse_topic(topic, subscribe_topic):
"""Parse an MQTT topic {sub_topic}/user/dev, return (user, dev) tuple.
@ -202,93 +153,6 @@ def _decrypt_payload(secret, topic, ciphertext):
return None
def context_from_config(async_see, config):
"""Create an async context from Home Assistant config."""
max_gps_accuracy = config.get(CONF_MAX_GPS_ACCURACY)
waypoint_import = config.get(CONF_WAYPOINT_IMPORT)
waypoint_whitelist = config.get(CONF_WAYPOINT_WHITELIST)
secret = config.get(CONF_SECRET)
region_mapping = config.get(CONF_REGION_MAPPING)
events_only = config.get(CONF_EVENTS_ONLY)
mqtt_topic = config.get(CONF_MQTT_TOPIC)
return OwnTracksContext(async_see, secret, max_gps_accuracy,
waypoint_import, waypoint_whitelist,
region_mapping, events_only, mqtt_topic)
class OwnTracksContext:
"""Hold the current OwnTracks context."""
def __init__(self, async_see, secret, max_gps_accuracy, import_waypoints,
waypoint_whitelist, region_mapping, events_only, mqtt_topic):
"""Initialize an OwnTracks context."""
self.async_see = async_see
self.secret = secret
self.max_gps_accuracy = max_gps_accuracy
self.mobile_beacons_active = defaultdict(set)
self.regions_entered = defaultdict(list)
self.import_waypoints = import_waypoints
self.waypoint_whitelist = waypoint_whitelist
self.region_mapping = region_mapping
self.events_only = events_only
self.mqtt_topic = mqtt_topic
@callback
def async_valid_accuracy(self, message):
"""Check if we should ignore this message."""
acc = message.get('acc')
if acc is None:
return False
try:
acc = float(acc)
except ValueError:
return False
if acc == 0:
_LOGGER.warning(
"Ignoring %s update because GPS accuracy is zero: %s",
message['_type'], message)
return False
if self.max_gps_accuracy is not None and \
acc > self.max_gps_accuracy:
_LOGGER.info("Ignoring %s update because expected GPS "
"accuracy %s is not met: %s",
message['_type'], self.max_gps_accuracy,
message)
return False
return True
async def async_see_beacons(self, hass, dev_id, kwargs_param):
"""Set active beacons to the current location."""
kwargs = kwargs_param.copy()
# Mobile beacons should always be set to the location of the
# tracking device. I get the device state and make the necessary
# changes to kwargs.
device_tracker_state = hass.states.get(
"device_tracker.{}".format(dev_id))
if device_tracker_state is not None:
acc = device_tracker_state.attributes.get("gps_accuracy")
lat = device_tracker_state.attributes.get("latitude")
lon = device_tracker_state.attributes.get("longitude")
kwargs['gps_accuracy'] = acc
kwargs['gps'] = (lat, lon)
# the battery state applies to the tracking device, not the beacon
# kwargs location is the beacon's configured lat/lon
kwargs.pop('battery', None)
for beacon in self.mobile_beacons_active[dev_id]:
kwargs['dev_id'] = "{}_{}".format(BEACON_DEV_ID, beacon)
kwargs['host_name'] = beacon
await self.async_see(**kwargs)
@HANDLERS.register('location')
async def async_handle_location_message(hass, context, message):
"""Handle a location message."""
@ -485,6 +349,8 @@ async def async_handle_message(hass, context, message):
"""Handle an OwnTracks message."""
msgtype = message.get('_type')
_LOGGER.debug("Received %s", message)
handler = HANDLERS.get(msgtype, async_handle_unsupported_msg)
await handler(hass, context, message)

82
homeassistant/components/device_tracker/owntracks_http.py
View File

@ -1,82 +0,0 @@
"""
Device tracker platform that adds support for OwnTracks over HTTP.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/device_tracker.owntracks_http/
"""
import json
import logging
import re
from aiohttp.web import Response
import voluptuous as vol
# pylint: disable=unused-import
from homeassistant.components.device_tracker.owntracks import ( # NOQA
PLATFORM_SCHEMA, REQUIREMENTS, async_handle_message, context_from_config)
from homeassistant.const import CONF_WEBHOOK_ID
import homeassistant.helpers.config_validation as cv
DEPENDENCIES = ['webhook']
_LOGGER = logging.getLogger(__name__)
EVENT_RECEIVED = 'owntracks_http_webhook_received'
EVENT_RESPONSE = 'owntracks_http_webhook_response_'
DOMAIN = 'device_tracker.owntracks_http'
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_WEBHOOK_ID): cv.string
})
async def async_setup_scanner(hass, config, async_see, discovery_info=None):
"""Set up OwnTracks HTTP component."""
context = context_from_config(async_see, config)
subscription = context.mqtt_topic
topic = re.sub('/#$', '', subscription)
async def handle_webhook(hass, webhook_id, request):
"""Handle webhook callback."""
headers = request.headers
data = dict()
if 'X-Limit-U' in headers:
data['user'] = headers['X-Limit-U']
elif 'u' in request.query:
data['user'] = request.query['u']
else:
return Response(
body=json.dumps({'error': 'You need to supply username.'}),
content_type="application/json"
)
if 'X-Limit-D' in headers:
data['device'] = headers['X-Limit-D']
elif 'd' in request.query:
data['device'] = request.query['d']
else:
return Response(
body=json.dumps({'error': 'You need to supply device name.'}),
content_type="application/json"
)
message = await request.json()
message['topic'] = '{}/{}/{}'.format(topic, data['user'],
data['device'])
try:
await async_handle_message(hass, context, message)
return Response(body=json.dumps([]), status=200,
content_type="application/json")
except ValueError:
_LOGGER.error("Received invalid JSON")
return None
hass.components.webhook.async_register(
'owntracks', 'OwnTracks', config['webhook_id'], handle_webhook)
return True

17
homeassistant/components/owntracks/.translations/en.json Normal file
View File

@ -0,0 +1,17 @@
{
"config": {
"abort": {
"one_instance_allowed": "Only a single instance is necessary."
},
"create_entry": {
"default": "\n\nOn Android, open [the OwnTracks app]({android_url}), go to preferences -> connection. Change the following settings:\n - Mode: Private HTTP\n - Host: {webhook_url}\n - Identification:\n - Username: `<Your name>`\n - Device ID: `<Your device name>`\n\nOn iOS, open [the OwnTracks app]({ios_url}), tap (i) icon in top left -> settings. Change the following settings:\n - Mode: HTTP\n - URL: {webhook_url}\n - Turn on authentication\n - UserID: `<Your name>`\n\n{secret}\n\nSee [the documentation]({docs_url}) for more information."
},
"step": {
"user": {
"description": "Are you sure you want to set up OwnTracks?",
"title": "Set up OwnTracks"
}
},
"title": "OwnTracks"
}
}

219
homeassistant/components/owntracks/__init__.py Normal file
View File

@ -0,0 +1,219 @@
"""Component for OwnTracks."""
from collections import defaultdict
import json
import logging
import re
from aiohttp.web import json_response
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_WEBHOOK_ID
from homeassistant.core import callback
from homeassistant.components import mqtt
from homeassistant.setup import async_when_setup
import homeassistant.helpers.config_validation as cv
from .config_flow import CONF_SECRET
DOMAIN = "owntracks"
REQUIREMENTS = ['libnacl==1.6.1']
DEPENDENCIES = ['device_tracker', 'webhook']
CONF_MAX_GPS_ACCURACY = 'max_gps_accuracy'
CONF_WAYPOINT_IMPORT = 'waypoints'
CONF_WAYPOINT_WHITELIST = 'waypoint_whitelist'
CONF_MQTT_TOPIC = 'mqtt_topic'
CONF_REGION_MAPPING = 'region_mapping'
CONF_EVENTS_ONLY = 'events_only'
BEACON_DEV_ID = 'beacon'
DEFAULT_OWNTRACKS_TOPIC = 'owntracks/#'
CONFIG_SCHEMA = vol.Schema({
vol.Optional(DOMAIN, default={}): {
vol.Optional(CONF_MAX_GPS_ACCURACY): vol.Coerce(float),
vol.Optional(CONF_WAYPOINT_IMPORT, default=True): cv.boolean,
vol.Optional(CONF_EVENTS_ONLY, default=False): cv.boolean,
vol.Optional(CONF_MQTT_TOPIC, default=DEFAULT_OWNTRACKS_TOPIC):
mqtt.valid_subscribe_topic,
vol.Optional(CONF_WAYPOINT_WHITELIST): vol.All(
cv.ensure_list, [cv.string]),
vol.Optional(CONF_SECRET): vol.Any(
vol.Schema({vol.Optional(cv.string): cv.string}),
cv.string),
vol.Optional(CONF_REGION_MAPPING, default={}): dict,
vol.Optional(CONF_WEBHOOK_ID): cv.string,
}
}, extra=vol.ALLOW_EXTRA)
_LOGGER = logging.getLogger(__name__)
async def async_setup(hass, config):
"""Initialize OwnTracks component."""
hass.data[DOMAIN] = {
'config': config[DOMAIN]
}
if not hass.config_entries.async_entries(DOMAIN):
hass.async_create_task(hass.config_entries.flow.async_init(
DOMAIN, context={'source': config_entries.SOURCE_IMPORT},
data={}
))
return True
async def async_setup_entry(hass, entry):
"""Set up OwnTracks entry."""
config = hass.data[DOMAIN]['config']
max_gps_accuracy = config.get(CONF_MAX_GPS_ACCURACY)
waypoint_import = config.get(CONF_WAYPOINT_IMPORT)
waypoint_whitelist = config.get(CONF_WAYPOINT_WHITELIST)
secret = config.get(CONF_SECRET) or entry.data[CONF_SECRET]
region_mapping = config.get(CONF_REGION_MAPPING)
events_only = config.get(CONF_EVENTS_ONLY)
mqtt_topic = config.get(CONF_MQTT_TOPIC)
context = OwnTracksContext(hass, secret, max_gps_accuracy,
waypoint_import, waypoint_whitelist,
region_mapping, events_only, mqtt_topic)
webhook_id = config.get(CONF_WEBHOOK_ID) or entry.data[CONF_WEBHOOK_ID]
hass.data[DOMAIN]['context'] = context
async_when_setup(hass, 'mqtt', async_connect_mqtt)
hass.components.webhook.async_register(
DOMAIN, 'OwnTracks', webhook_id, handle_webhook)
hass.async_create_task(hass.config_entries.async_forward_entry_setup(
entry, 'device_tracker'))
return True
async def async_connect_mqtt(hass, component):
"""Subscribe to MQTT topic."""
context = hass.data[DOMAIN]['context']
async def async_handle_mqtt_message(topic, payload, qos):
"""Handle incoming OwnTracks message."""
try:
message = json.loads(payload)
except ValueError:
# If invalid JSON
_LOGGER.error("Unable to parse payload as JSON: %s", payload)
return
message['topic'] = topic
hass.helpers.dispatcher.async_dispatcher_send(
DOMAIN, hass, context, message)
await hass.components.mqtt.async_subscribe(
context.mqtt_topic, async_handle_mqtt_message, 1)
return True
async def handle_webhook(hass, webhook_id, request):
"""Handle webhook callback."""
context = hass.data[DOMAIN]['context']
message = await request.json()
# Android doesn't populate topic
if 'topic' not in message:
headers = request.headers
user = headers.get('X-Limit-U')
device = headers.get('X-Limit-D', user)
if user is None:
_LOGGER.warning('Set a username in Connection -> Identification')
return json_response(
{'error': 'You need to supply username.'},
status=400
)
topic_base = re.sub('/#$', '', context.mqtt_topic)
message['topic'] = '{}/{}/{}'.format(topic_base, user, device)
hass.helpers.dispatcher.async_dispatcher_send(
DOMAIN, hass, context, message)
return json_response([])
class OwnTracksContext:
"""Hold the current OwnTracks context."""
def __init__(self, hass, secret, max_gps_accuracy, import_waypoints,
waypoint_whitelist, region_mapping, events_only, mqtt_topic):
"""Initialize an OwnTracks context."""
self.hass = hass
self.secret = secret
self.max_gps_accuracy = max_gps_accuracy
self.mobile_beacons_active = defaultdict(set)
self.regions_entered = defaultdict(list)
self.import_waypoints = import_waypoints
self.waypoint_whitelist = waypoint_whitelist
self.region_mapping = region_mapping
self.events_only = events_only
self.mqtt_topic = mqtt_topic
@callback
def async_valid_accuracy(self, message):
"""Check if we should ignore this message."""
acc = message.get('acc')
if acc is None:
return False
try:
acc = float(acc)
except ValueError:
return False
if acc == 0:
_LOGGER.warning(
"Ignoring %s update because GPS accuracy is zero: %s",
message['_type'], message)
return False
if self.max_gps_accuracy is not None and \
acc > self.max_gps_accuracy:
_LOGGER.info("Ignoring %s update because expected GPS "
"accuracy %s is not met: %s",
message['_type'], self.max_gps_accuracy,
message)
return False
return True
async def async_see(self, **data):
"""Send a see message to the device tracker."""
await self.hass.components.device_tracker.async_see(**data)
async def async_see_beacons(self, hass, dev_id, kwargs_param):
"""Set active beacons to the current location."""
kwargs = kwargs_param.copy()
# Mobile beacons should always be set to the location of the
# tracking device. I get the device state and make the necessary
# changes to kwargs.
device_tracker_state = hass.states.get(
"device_tracker.{}".format(dev_id))
if device_tracker_state is not None:
acc = device_tracker_state.attributes.get("gps_accuracy")
lat = device_tracker_state.attributes.get("latitude")
lon = device_tracker_state.attributes.get("longitude")
kwargs['gps_accuracy'] = acc
kwargs['gps'] = (lat, lon)
# the battery state applies to the tracking device, not the beacon
# kwargs location is the beacon's configured lat/lon
kwargs.pop('battery', None)
for beacon in self.mobile_beacons_active[dev_id]:
kwargs['dev_id'] = "{}_{}".format(BEACON_DEV_ID, beacon)
kwargs['host_name'] = beacon
await self.async_see(**kwargs)

79
homeassistant/components/owntracks/config_flow.py Normal file
View File

@ -0,0 +1,79 @@
"""Config flow for OwnTracks."""
from homeassistant import config_entries
from homeassistant.const import CONF_WEBHOOK_ID
from homeassistant.auth.util import generate_secret
CONF_SECRET = 'secret'
def supports_encryption():
"""Test if we support encryption."""
try:
# pylint: disable=unused-variable
import libnacl # noqa
return True
except OSError:
return False
@config_entries.HANDLERS.register('owntracks')
class OwnTracksFlow(config_entries.ConfigFlow):
"""Set up OwnTracks."""
VERSION = 1
async def async_step_user(self, user_input=None):
"""Handle a user initiated set up flow to create OwnTracks webhook."""
if self._async_current_entries():
return self.async_abort(reason='one_instance_allowed')
if user_input is None:
return self.async_show_form(
step_id='user',
)
webhook_id = self.hass.components.webhook.async_generate_id()
webhook_url = \
self.hass.components.webhook.async_generate_url(webhook_id)
secret = generate_secret(16)
if supports_encryption():
secret_desc = (
"The encryption key is {secret} "
"(on Android under preferences -> advanced)")
else:
secret_desc = (
"Encryption is not supported because libsodium is not "
"installed.")
return self.async_create_entry(
title="OwnTracks",
data={
CONF_WEBHOOK_ID: webhook_id,
CONF_SECRET: secret
},
description_placeholders={
'secret': secret_desc,
'webhook_url': webhook_url,
'android_url':
'https://play.google.com/store/apps/details?'
'id=org.owntracks.android',
'ios_url':
'https://itunes.apple.com/us/app/owntracks/id692424691?mt=8',
'docs_url':
'https://www.home-assistant.io/components/owntracks/'
}
)
async def async_step_import(self, user_input):
"""Import a config flow from configuration."""
webhook_id = self.hass.components.webhook.async_generate_id()
secret = generate_secret(16)
return self.async_create_entry(
title="OwnTracks",
data={
CONF_WEBHOOK_ID: webhook_id,
CONF_SECRET: secret
}
)

17
homeassistant/components/owntracks/strings.json Normal file
View File

@ -0,0 +1,17 @@
{
"config": {
"title": "OwnTracks",
"step": {
"user": {
"title": "Set up OwnTracks",
"description": "Are you sure you want to set up OwnTracks?"
}
},
"abort": {
"one_instance_allowed": "Only a single instance is necessary."
},
"create_entry": {
"default": "\n\nOn Android, open [the OwnTracks app]({android_url}), go to preferences -> connection. Change the following settings:\n - Mode: Private HTTP\n - Host: {webhook_url}\n - Identification:\n - Username: `<Your name>`\n - Device ID: `<Your device name>`\n\nOn iOS, open [the OwnTracks app]({ios_url}), tap (i) icon in top left -> settings. Change the following settings:\n - Mode: HTTP\n - URL: {webhook_url}\n - Turn on authentication\n - UserID: `<Your name>`\n\n{secret}\n\nSee [the documentation]({docs_url}) for more information."
}
}
}

1
homeassistant/config_entries.py
View File

@ -149,6 +149,7 @@ FLOWS = [
'mqtt',
'nest',
'openuv',
'owntracks',
'point',
'rainmachine',
'simplisafe',

34
homeassistant/setup.py
View File

@ -4,7 +4,7 @@ import logging.handlers
from timeit import default_timer as timer
from types import ModuleType
from typing import Optional, Dict, List
from typing import Awaitable, Callable, Optional, Dict, List
from homeassistant import requirements, core, loader, config as conf_util
from homeassistant.config import async_notify_setup_error
@ -248,3 +248,35 @@ async def async_process_deps_reqs(
raise HomeAssistantError("Could not install all requirements.")
processed.add(name)
@core.callback
def async_when_setup(
hass: core.HomeAssistant, component: str,
when_setup_cb: Callable[
[core.HomeAssistant, str], Awaitable[None]]) -> None:
"""Call a method when a component is setup."""
async def when_setup() -> None:
"""Call the callback."""
try:
await when_setup_cb(hass, component)
except Exception: # pylint: disable=broad-except
_LOGGER.exception('Error handling when_setup callback for %s',
component)
# Running it in a new task so that it always runs after
if component in hass.config.components:
hass.async_create_task(when_setup())
return
unsub = None
async def loaded_event(event: core.Event) -> None:
"""Call the callback."""
if event.data[ATTR_COMPONENT] != component:
return
unsub() # type: ignore
await when_setup()
unsub = hass.bus.async_listen(EVENT_COMPONENT_LOADED, loaded_event)

3
requirements_all.txt
View File

@ -559,8 +559,7 @@ konnected==0.1.4
# homeassistant.components.eufy
lakeside==0.10
# homeassistant.components.device_tracker.owntracks
# homeassistant.components.device_tracker.owntracks_http
# homeassistant.components.owntracks
libnacl==1.6.1
# homeassistant.components.dyson

154
tests/components/device_tracker/test_owntracks.py
View File

@ -4,12 +4,11 @@ from asynctest import patch
import pytest
from tests.common import (
assert_setup_component, async_fire_mqtt_message, mock_coro, mock_component,
async_mock_mqtt_component)
import homeassistant.components.device_tracker.owntracks as owntracks
async_fire_mqtt_message, mock_coro, mock_component,
async_mock_mqtt_component, MockConfigEntry)
from homeassistant.components import owntracks
from homeassistant.setup import async_setup_component
from homeassistant.components import device_tracker
from homeassistant.const import CONF_PLATFORM, STATE_NOT_HOME
from homeassistant.const import STATE_NOT_HOME
USER = 'greg'
DEVICE = 'phone'
@ -290,6 +289,25 @@ def setup_comp(hass):
'zone.outer', 'zoning', OUTER_ZONE)
async def setup_owntracks(hass, config,
ctx_cls=owntracks.OwnTracksContext):
"""Set up OwnTracks."""
await async_mock_mqtt_component(hass)
MockConfigEntry(domain='owntracks', data={
'webhook_id': 'owntracks_test',
'secret': 'abcd',
}).add_to_hass(hass)
with patch('homeassistant.components.device_tracker.async_load_config',
return_value=mock_coro([])), \
patch('homeassistant.components.device_tracker.'
'load_yaml_config_file', return_value=mock_coro({})), \
patch.object(owntracks, 'OwnTracksContext', ctx_cls):
assert await async_setup_component(
hass, 'owntracks', {'owntracks': config})
@pytest.fixture
def context(hass, setup_comp):
"""Set up the mocked context."""
@ -306,20 +324,11 @@ def context(hass, setup_comp):
context = orig_context(*args)
return context
with patch('homeassistant.components.device_tracker.async_load_config',
return_value=mock_coro([])), \
patch('homeassistant.components.device_tracker.'
'load_yaml_config_file', return_value=mock_coro({})), \
patch.object(owntracks, 'OwnTracksContext', store_context), \
assert_setup_component(1, device_tracker.DOMAIN):
assert hass.loop.run_until_complete(async_setup_component(
hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_MAX_GPS_ACCURACY: 200,
CONF_WAYPOINT_IMPORT: True,
CONF_WAYPOINT_WHITELIST: ['jon', 'greg']
}}))
hass.loop.run_until_complete(setup_owntracks(hass, {
CONF_MAX_GPS_ACCURACY: 200,
CONF_WAYPOINT_IMPORT: True,
CONF_WAYPOINT_WHITELIST: ['jon', 'greg']
}, store_context))
def get_context():
"""Get the current context."""
@ -1211,19 +1220,14 @@ async def test_waypoint_import_blacklist(hass, context):
assert wayp is None
async def test_waypoint_import_no_whitelist(hass, context):
async def test_waypoint_import_no_whitelist(hass, config_context):
"""Test import of list of waypoints with no whitelist set."""
async def mock_see(**kwargs):
"""Fake see method for owntracks."""
return
test_config = {
CONF_PLATFORM: 'owntracks',
await setup_owntracks(hass, {
CONF_MAX_GPS_ACCURACY: 200,
CONF_WAYPOINT_IMPORT: True,
CONF_MQTT_TOPIC: 'owntracks/#',
}
await owntracks.async_setup_scanner(hass, test_config, mock_see)
})
waypoints_message = WAYPOINTS_EXPORTED_MESSAGE.copy()
await send_message(hass, WAYPOINTS_TOPIC_BLOCKED, waypoints_message)
# Check if it made it into states
@ -1364,12 +1368,9 @@ def config_context(hass, setup_comp):
mock_cipher)
async def test_encrypted_payload(hass, config_context):
"""Test encrypted payload."""
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_SECRET: TEST_SECRET_KEY,
}})
await setup_owntracks(hass, {
CONF_SECRET: TEST_SECRET_KEY,
})
await send_message(hass, LOCATION_TOPIC, MOCK_ENCRYPTED_LOCATION_MESSAGE)
assert_location_latitude(hass, LOCATION_MESSAGE['lat'])
@ -1378,13 +1379,11 @@ async def test_encrypted_payload(hass, config_context):
mock_cipher)
async def test_encrypted_payload_topic_key(hass, config_context):
"""Test encrypted payload with a topic key."""
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_SECRET: {
LOCATION_TOPIC: TEST_SECRET_KEY,
}}})
await setup_owntracks(hass, {
CONF_SECRET: {
LOCATION_TOPIC: TEST_SECRET_KEY,
}
})
await send_message(hass, LOCATION_TOPIC, MOCK_ENCRYPTED_LOCATION_MESSAGE)
assert_location_latitude(hass, LOCATION_MESSAGE['lat'])
@ -1394,12 +1393,10 @@ async def test_encrypted_payload_topic_key(hass, config_context):
async def test_encrypted_payload_no_key(hass, config_context):
"""Test encrypted payload with no key, ."""
assert hass.states.get(DEVICE_TRACKER_STATE) is None
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
# key missing
}})
await setup_owntracks(hass, {
CONF_SECRET: {
}
})
await send_message(hass, LOCATION_TOPIC, MOCK_ENCRYPTED_LOCATION_MESSAGE)
assert hass.states.get(DEVICE_TRACKER_STATE) is None
@ -1408,12 +1405,9 @@ async def test_encrypted_payload_no_key(hass, config_context):
mock_cipher)
async def test_encrypted_payload_wrong_key(hass, config_context):
"""Test encrypted payload with wrong key."""
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_SECRET: 'wrong key',
}})
await setup_owntracks(hass, {
CONF_SECRET: 'wrong key',
})
await send_message(hass, LOCATION_TOPIC, MOCK_ENCRYPTED_LOCATION_MESSAGE)
assert hass.states.get(DEVICE_TRACKER_STATE) is None
@ -1422,13 +1416,11 @@ async def test_encrypted_payload_wrong_key(hass, config_context):
mock_cipher)
async def test_encrypted_payload_wrong_topic_key(hass, config_context):
"""Test encrypted payload with wrong topic key."""
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_SECRET: {
LOCATION_TOPIC: 'wrong key'
}}})
await setup_owntracks(hass, {
CONF_SECRET: {
LOCATION_TOPIC: 'wrong key'
},
})
await send_message(hass, LOCATION_TOPIC, MOCK_ENCRYPTED_LOCATION_MESSAGE)
assert hass.states.get(DEVICE_TRACKER_STATE) is None
@ -1437,13 +1429,10 @@ async def test_encrypted_payload_wrong_topic_key(hass, config_context):
mock_cipher)
async def test_encrypted_payload_no_topic_key(hass, config_context):
"""Test encrypted payload with no topic key."""
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_SECRET: {
'owntracks/{}/{}'.format(USER, 'otherdevice'): 'foobar'
}}})
await setup_owntracks(hass, {
CONF_SECRET: {
'owntracks/{}/{}'.format(USER, 'otherdevice'): 'foobar'
}})
await send_message(hass, LOCATION_TOPIC, MOCK_ENCRYPTED_LOCATION_MESSAGE)
assert hass.states.get(DEVICE_TRACKER_STATE) is None
@ -1456,12 +1445,9 @@ async def test_encrypted_payload_libsodium(hass, config_context):
pytest.skip("libnacl/libsodium is not installed")
return
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_SECRET: TEST_SECRET_KEY,
}})
await setup_owntracks(hass, {
CONF_SECRET: TEST_SECRET_KEY,
})
await send_message(hass, LOCATION_TOPIC, ENCRYPTED_LOCATION_MESSAGE)
assert_location_latitude(hass, LOCATION_MESSAGE['lat'])
@ -1469,12 +1455,9 @@ async def test_encrypted_payload_libsodium(hass, config_context):
async def test_customized_mqtt_topic(hass, config_context):
"""Test subscribing to a custom mqtt topic."""
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_MQTT_TOPIC: 'mytracks/#',
}})
await setup_owntracks(hass, {
CONF_MQTT_TOPIC: 'mytracks/#',
})
topic = 'mytracks/{}/{}'.format(USER, DEVICE)
@ -1484,14 +1467,11 @@ async def test_customized_mqtt_topic(hass, config_context):
async def test_region_mapping(hass, config_context):
"""Test region to zone mapping."""
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(hass, device_tracker.DOMAIN, {
device_tracker.DOMAIN: {
CONF_PLATFORM: 'owntracks',
CONF_REGION_MAPPING: {
'foo': 'inner'
},
}})
await setup_owntracks(hass, {
CONF_REGION_MAPPING: {
'foo': 'inner'
},
})
hass.states.async_set(
'zone.inner', 'zoning', INNER_ZONE)

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

@ -0,0 +1 @@
"""Tests for OwnTracks component."""

1
tests/components/owntracks/test_config_flow.py Normal file
View File

@ -0,0 +1 @@
"""Tests for OwnTracks config flow."""

97
tests/components/device_tracker/test_owntracks_http.py → tests/components/owntracks/test_init.py
View File

@ -1,14 +1,11 @@
"""Test the owntracks_http platform."""
import asyncio
from unittest.mock import patch
import os
import pytest
from homeassistant.components import device_tracker
from homeassistant.setup import async_setup_component
from tests.common import mock_component, mock_coro
from tests.common import mock_component, MockConfigEntry
MINIMAL_LOCATION_MESSAGE = {
'_type': 'location',
@ -36,38 +33,33 @@ LOCATION_MESSAGE = {
}
@pytest.fixture(autouse=True)
def owntracks_http_cleanup(hass):
"""Remove known_devices.yaml."""
try:
os.remove(hass.config.path(device_tracker.YAML_DEVICES))
except OSError:
pass
@pytest.fixture
def mock_client(hass, aiohttp_client):
"""Start the Hass HTTP component."""
mock_component(hass, 'group')
mock_component(hass, 'zone')
with patch('homeassistant.components.device_tracker.async_load_config',
return_value=mock_coro([])):
hass.loop.run_until_complete(
async_setup_component(hass, 'device_tracker', {
'device_tracker': {
'platform': 'owntracks_http',
'webhook_id': 'owntracks_test'
}
}))
mock_component(hass, 'device_tracker')
MockConfigEntry(domain='owntracks', data={
'webhook_id': 'owntracks_test',
'secret': 'abcd',
}).add_to_hass(hass)
hass.loop.run_until_complete(async_setup_component(hass, 'owntracks', {}))
return hass.loop.run_until_complete(aiohttp_client(hass.http.app))
@asyncio.coroutine
def test_handle_valid_message(mock_client):
"""Test that we forward messages correctly to OwnTracks."""
resp = yield from mock_client.post('/api/webhook/owntracks_test?'
'u=test&d=test',
json=LOCATION_MESSAGE)
resp = yield from mock_client.post(
'/api/webhook/owntracks_test',
json=LOCATION_MESSAGE,
headers={
'X-Limit-u': 'Paulus',
'X-Limit-d': 'Pixel',
}
)
assert resp.status == 200
@ -78,9 +70,14 @@ def test_handle_valid_message(mock_client):
@asyncio.coroutine
def test_handle_valid_minimal_message(mock_client):
"""Test that we forward messages correctly to OwnTracks."""
resp = yield from mock_client.post('/api/webhook/owntracks_test?'
'u=test&d=test',
json=MINIMAL_LOCATION_MESSAGE)
resp = yield from mock_client.post(
'/api/webhook/owntracks_test',
json=MINIMAL_LOCATION_MESSAGE,
headers={
'X-Limit-u': 'Paulus',
'X-Limit-d': 'Pixel',
}
)
assert resp.status == 200
@ -91,8 +88,14 @@ def test_handle_valid_minimal_message(mock_client):
@asyncio.coroutine
def test_handle_value_error(mock_client):
"""Test we don't disclose that this is a valid webhook."""
resp = yield from mock_client.post('/api/webhook/owntracks_test'
'?u=test&d=test', json='')
resp = yield from mock_client.post(
'/api/webhook/owntracks_test',
json='',
headers={
'X-Limit-u': 'Paulus',
'X-Limit-d': 'Pixel',
}
)
assert resp.status == 200
@ -103,10 +106,15 @@ def test_handle_value_error(mock_client):
@asyncio.coroutine
def test_returns_error_missing_username(mock_client):
"""Test that an error is returned when username is missing."""
resp = yield from mock_client.post('/api/webhook/owntracks_test?d=test',
json=LOCATION_MESSAGE)
resp = yield from mock_client.post(
'/api/webhook/owntracks_test',
json=LOCATION_MESSAGE,
headers={
'X-Limit-d': 'Pixel',
}
)
assert resp.status == 200
assert resp.status == 400
json = yield from resp.json()
assert json == {'error': 'You need to supply username.'}
@ -115,10 +123,27 @@ def test_returns_error_missing_username(mock_client):
@asyncio.coroutine
def test_returns_error_missing_device(mock_client):
"""Test that an error is returned when device name is missing."""
resp = yield from mock_client.post('/api/webhook/owntracks_test?u=test',
json=LOCATION_MESSAGE)
resp = yield from mock_client.post(
'/api/webhook/owntracks_test',
json=LOCATION_MESSAGE,
headers={
'X-Limit-u': 'Paulus',
}
)
assert resp.status == 200
json = yield from resp.json()
assert json == {'error': 'You need to supply device name.'}
assert json == []
async def test_config_flow_import(hass):
"""Test that we automatically create a config flow."""
assert not hass.config_entries.async_entries('owntracks')
assert await async_setup_component(hass, 'owntracks', {
'owntracks': {
}
})
await hass.async_block_till_done()
assert hass.config_entries.async_entries('owntracks')

35
tests/test_setup.py
View File

@ -9,7 +9,8 @@ import logging
import voluptuous as vol
from homeassistant.core import callback
from homeassistant.const import EVENT_HOMEASSISTANT_START
from homeassistant.const import (
EVENT_HOMEASSISTANT_START, EVENT_COMPONENT_LOADED)
import homeassistant.config as config_util
from homeassistant import setup, loader
import homeassistant.util.dt as dt_util
@ -459,3 +460,35 @@ def test_platform_no_warn_slow(hass):
hass, 'test_component1', {})
assert result
assert not mock_call.called
async def test_when_setup_already_loaded(hass):
"""Test when setup."""
calls = []
async def mock_callback(hass, component):
"""Mock callback."""
calls.append(component)
setup.async_when_setup(hass, 'test', mock_callback)
await hass.async_block_till_done()
assert calls == []
hass.config.components.add('test')
hass.bus.async_fire(EVENT_COMPONENT_LOADED, {
'component': 'test'
})
await hass.async_block_till_done()
assert calls == ['test']
# Event listener should be gone
hass.bus.async_fire(EVENT_COMPONENT_LOADED, {
'component': 'test'
})
await hass.async_block_till_done()
assert calls == ['test']
# Should be called right away
setup.async_when_setup(hass, 'test', mock_callback)
await hass.async_block_till_done()
assert calls == ['test', 'test']