Clean up owntracks tests (#21378)

This commit is contained in:
Martin Hjelmare 2019-02-24 02:17:49 +01:00 committed by Andrew Sayre
parent fc13e37d8d
commit d44269981b

View File

@ -1,14 +1,16 @@
"""The tests for the Owntracks device tracker.""" """The tests for the Owntracks device tracker."""
import json import json
from asynctest import patch from asynctest import patch
import pytest import pytest
from tests.common import (
async_fire_mqtt_message, mock_coro, mock_component,
async_mock_mqtt_component, MockConfigEntry)
from homeassistant.components import owntracks from homeassistant.components import owntracks
from homeassistant.setup import async_setup_component
from homeassistant.const import STATE_NOT_HOME from homeassistant.const import STATE_NOT_HOME
from homeassistant.setup import async_setup_component
from tests.common import (
MockConfigEntry, async_fire_mqtt_message, async_mock_mqtt_component,
mock_coro)
USER = 'greg' USER = 'greg'
DEVICE = 'phone' DEVICE = 'phone'
@ -45,8 +47,8 @@ FIVE_M = TEST_ZONE_DEG_PER_M * 5.0
# Home Assistant Zones # Home Assistant Zones
INNER_ZONE = { INNER_ZONE = {
'name': 'zone', 'name': 'zone',
'latitude': TEST_ZONE_LAT+0.1, 'latitude': TEST_ZONE_LAT + 0.1,
'longitude': TEST_ZONE_LON+0.1, 'longitude': TEST_ZONE_LON + 0.1,
'radius': 50 'radius': 50
} }
@ -271,12 +273,14 @@ BAD_MESSAGE = {
BAD_JSON_PREFIX = '--$this is bad json#--' BAD_JSON_PREFIX = '--$this is bad json#--'
BAD_JSON_SUFFIX = '** and it ends here ^^' BAD_JSON_SUFFIX = '** and it ends here ^^'
# pylint: disable=invalid-name, len-as-condition, redefined-outer-name
@pytest.fixture @pytest.fixture
def setup_comp(hass): def setup_comp(hass, mock_device_tracker_conf):
"""Initialize components.""" """Initialize components."""
mock_component(hass, 'group') assert hass.loop.run_until_complete(async_setup_component(
mock_component(hass, 'zone') hass, 'persistent_notification', {}))
hass.loop.run_until_complete(async_setup_component( hass.loop.run_until_complete(async_setup_component(
hass, 'device_tracker', {})) hass, 'device_tracker', {}))
hass.loop.run_until_complete(async_mock_mqtt_component(hass)) hass.loop.run_until_complete(async_mock_mqtt_component(hass))
@ -289,48 +293,42 @@ def setup_comp(hass):
hass.states.async_set( hass.states.async_set(
'zone.outer', 'zoning', OUTER_ZONE) 'zone.outer', 'zoning', OUTER_ZONE)
yield
async def setup_owntracks(hass, config, async def setup_owntracks(hass, config,
ctx_cls=owntracks.OwnTracksContext): ctx_cls=owntracks.OwnTracksContext):
"""Set up OwnTracks.""" """Set up OwnTracks."""
await async_mock_mqtt_component(hass)
MockConfigEntry(domain='owntracks', data={ MockConfigEntry(domain='owntracks', data={
'webhook_id': 'owntracks_test', 'webhook_id': 'owntracks_test',
'secret': 'abcd', 'secret': 'abcd',
}).add_to_hass(hass) }).add_to_hass(hass)
with patch('homeassistant.components.device_tracker.async_load_config', with patch.object(owntracks, 'OwnTracksContext', ctx_cls):
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( assert await async_setup_component(
hass, 'owntracks', {'owntracks': config}) hass, 'owntracks', {'owntracks': config})
await hass.async_block_till_done()
@pytest.fixture @pytest.fixture
def context(hass, setup_comp): def context(hass, setup_comp):
"""Set up the mocked context.""" """Set up the mocked context."""
patcher = patch('homeassistant.components.device_tracker.'
'DeviceTracker.async_update_config')
patcher.start()
orig_context = owntracks.OwnTracksContext orig_context = owntracks.OwnTracksContext
context = None context = None
# pylint: disable=no-value-for-parameter
def store_context(*args): def store_context(*args):
"""Store the context."""
nonlocal context nonlocal context
context = orig_context(*args) context = orig_context(*args)
return context return context
hass.loop.run_until_complete(setup_owntracks(hass, { hass.loop.run_until_complete(setup_owntracks(hass, {
CONF_MAX_GPS_ACCURACY: 200, CONF_MAX_GPS_ACCURACY: 200,
CONF_WAYPOINT_IMPORT: True, CONF_WAYPOINT_IMPORT: True,
CONF_WAYPOINT_WHITELIST: ['jon', 'greg'] CONF_WAYPOINT_WHITELIST: ['jon', 'greg']
}, store_context)) }, store_context))
def get_context(): def get_context():
"""Get the current context.""" """Get the current context."""
@ -338,8 +336,6 @@ def context(hass, setup_comp):
yield get_context yield get_context
patcher.stop()
async def send_message(hass, topic, message, corrupt=False): async def send_message(hass, topic, message, corrupt=False):
"""Test the sending of a message.""" """Test the sending of a message."""
@ -851,7 +847,7 @@ async def test_event_beacon_unknown_zone_no_location(hass, context):
# that will be tracked at my current location. Except # that will be tracked at my current location. Except
# in this case my Device hasn't had a location message # in this case my Device hasn't had a location message
# yet so it's in an odd state where it has state.state # yet so it's in an odd state where it has state.state
# None and no GPS coords so set the beacon to. # None and no GPS coords to set the beacon to.
hass.states.async_set(DEVICE_TRACKER_STATE, None) hass.states.async_set(DEVICE_TRACKER_STATE, None)
message = build_message( message = build_message(
@ -993,8 +989,7 @@ async def test_mobile_multiple_async_enter_exit(hass, context):
await hass.async_block_till_done() await hass.async_block_till_done()
await send_message(hass, EVENT_TOPIC, MOBILE_BEACON_LEAVE_EVENT_MESSAGE) await send_message(hass, EVENT_TOPIC, MOBILE_BEACON_LEAVE_EVENT_MESSAGE)
assert len(context().mobile_beacons_active['greg_phone']) == \ assert len(context().mobile_beacons_active['greg_phone']) == 0
0
async def test_mobile_multiple_enter_exit(hass, context): async def test_mobile_multiple_enter_exit(hass, context):
@ -1003,8 +998,7 @@ async def test_mobile_multiple_enter_exit(hass, context):
await send_message(hass, EVENT_TOPIC, MOBILE_BEACON_ENTER_EVENT_MESSAGE) await send_message(hass, EVENT_TOPIC, MOBILE_BEACON_ENTER_EVENT_MESSAGE)
await send_message(hass, EVENT_TOPIC, MOBILE_BEACON_LEAVE_EVENT_MESSAGE) await send_message(hass, EVENT_TOPIC, MOBILE_BEACON_LEAVE_EVENT_MESSAGE)
assert len(context().mobile_beacons_active['greg_phone']) == \ assert len(context().mobile_beacons_active['greg_phone']) == 0
0
async def test_complex_movement(hass, context): async def test_complex_movement(hass, context):
@ -1153,38 +1147,46 @@ async def test_complex_movement_sticky_keys_beacon(hass, context):
# leave keys # leave keys
await send_message(hass, LOCATION_TOPIC, location_message) await send_message(hass, LOCATION_TOPIC, location_message)
await send_message(hass, EVENT_TOPIC, MOBILE_BEACON_LEAVE_EVENT_MESSAGE) await send_message(hass, EVENT_TOPIC, MOBILE_BEACON_LEAVE_EVENT_MESSAGE)
assert_location_latitude(hass, INNER_ZONE['latitude'])
assert_location_state(hass, 'inner') assert_location_state(hass, 'inner')
assert_mobile_tracker_state(hass, 'inner') assert_mobile_tracker_state(hass, 'inner')
assert_mobile_tracker_latitude(hass, INNER_ZONE['latitude'])
# leave inner region beacon # leave inner region beacon
await send_message(hass, EVENT_TOPIC, REGION_BEACON_LEAVE_MESSAGE) await send_message(hass, EVENT_TOPIC, REGION_BEACON_LEAVE_MESSAGE)
await send_message(hass, LOCATION_TOPIC, location_message) await send_message(hass, LOCATION_TOPIC, location_message)
assert_location_state(hass, 'inner') assert_location_state(hass, 'inner')
assert_mobile_tracker_state(hass, 'inner') assert_mobile_tracker_state(hass, 'inner')
assert_mobile_tracker_latitude(hass, INNER_ZONE['latitude'])
# enter inner region beacon # enter inner region beacon
await send_message(hass, EVENT_TOPIC, REGION_BEACON_ENTER_MESSAGE) await send_message(hass, EVENT_TOPIC, REGION_BEACON_ENTER_MESSAGE)
await send_message(hass, LOCATION_TOPIC, location_message) await send_message(hass, LOCATION_TOPIC, location_message)
assert_location_latitude(hass, INNER_ZONE['latitude']) assert_location_latitude(hass, INNER_ZONE['latitude'])
assert_location_state(hass, 'inner') assert_location_state(hass, 'inner')
assert_mobile_tracker_state(hass, 'inner')
assert_mobile_tracker_latitude(hass, INNER_ZONE['latitude'])
# enter keys # enter keys
await send_message(hass, EVENT_TOPIC, MOBILE_BEACON_ENTER_EVENT_MESSAGE) await send_message(hass, EVENT_TOPIC, MOBILE_BEACON_ENTER_EVENT_MESSAGE)
await send_message(hass, LOCATION_TOPIC, location_message) await send_message(hass, LOCATION_TOPIC, location_message)
assert_location_state(hass, 'inner') assert_location_state(hass, 'inner')
assert_mobile_tracker_state(hass, 'inner') assert_mobile_tracker_state(hass, 'inner')
assert_mobile_tracker_latitude(hass, INNER_ZONE['latitude'])
# leave keys # leave keys
await send_message(hass, LOCATION_TOPIC, location_message) await send_message(hass, LOCATION_TOPIC, location_message)
await send_message(hass, EVENT_TOPIC, MOBILE_BEACON_LEAVE_EVENT_MESSAGE) await send_message(hass, EVENT_TOPIC, MOBILE_BEACON_LEAVE_EVENT_MESSAGE)
assert_location_state(hass, 'inner') assert_location_state(hass, 'inner')
assert_mobile_tracker_state(hass, 'inner') assert_mobile_tracker_state(hass, 'inner')
assert_mobile_tracker_latitude(hass, INNER_ZONE['latitude'])
# leave inner region beacon # leave inner region beacon
await send_message(hass, EVENT_TOPIC, REGION_BEACON_LEAVE_MESSAGE) await send_message(hass, EVENT_TOPIC, REGION_BEACON_LEAVE_MESSAGE)
await send_message(hass, LOCATION_TOPIC, location_message) await send_message(hass, LOCATION_TOPIC, location_message)
assert_location_state(hass, 'inner') assert_location_state(hass, 'inner')
assert_mobile_tracker_state(hass, 'inner') assert_mobile_tracker_state(hass, 'inner')
assert_mobile_tracker_latitude(hass, INNER_ZONE['latitude'])
# GPS leave inner region, I'm in the 'outer' region now # GPS leave inner region, I'm in the 'outer' region now
# but on GPS coords # but on GPS coords
@ -1222,7 +1224,7 @@ async def test_waypoint_import_blacklist(hass, context):
assert wayp is None assert wayp is None
async def test_waypoint_import_no_whitelist(hass, config_context): async def test_waypoint_import_no_whitelist(hass, setup_comp):
"""Test import of list of waypoints with no whitelist set.""" """Test import of list of waypoints with no whitelist set."""
await setup_owntracks(hass, { await setup_owntracks(hass, {
CONF_MAX_GPS_ACCURACY: 200, CONF_MAX_GPS_ACCURACY: 200,
@ -1296,7 +1298,6 @@ def generate_ciphers(secret):
# libnacl ciphertext generation will fail if the module # libnacl ciphertext generation will fail if the module
# cannot be imported. However, the test for decryption # cannot be imported. However, the test for decryption
# also relies on this library and won't be run without it. # also relies on this library and won't be run without it.
import json
import pickle import pickle
import base64 import base64
@ -1304,9 +1305,8 @@ def generate_ciphers(secret):
from libnacl import crypto_secretbox_KEYBYTES as KEYLEN from libnacl import crypto_secretbox_KEYBYTES as KEYLEN
from libnacl.secret import SecretBox from libnacl.secret import SecretBox
key = secret.encode("utf-8")[:KEYLEN].ljust(KEYLEN, b'\0') key = secret.encode("utf-8")[:KEYLEN].ljust(KEYLEN, b'\0')
ctxt = base64.b64encode(SecretBox(key).encrypt( ctxt = base64.b64encode(SecretBox(key).encrypt(json.dumps(
json.dumps(DEFAULT_LOCATION_MESSAGE).encode("utf-8")) DEFAULT_LOCATION_MESSAGE).encode("utf-8"))).decode("utf-8")
).decode("utf-8")
except (ImportError, OSError): except (ImportError, OSError):
ctxt = '' ctxt = ''
@ -1368,7 +1368,7 @@ def config_context(hass, setup_comp):
@patch('homeassistant.components.owntracks.device_tracker.get_cipher', @patch('homeassistant.components.owntracks.device_tracker.get_cipher',
mock_cipher) mock_cipher)
async def test_encrypted_payload(hass, config_context): async def test_encrypted_payload(hass, setup_comp):
"""Test encrypted payload.""" """Test encrypted payload."""
await setup_owntracks(hass, { await setup_owntracks(hass, {
CONF_SECRET: TEST_SECRET_KEY, CONF_SECRET: TEST_SECRET_KEY,
@ -1379,7 +1379,7 @@ async def test_encrypted_payload(hass, config_context):
@patch('homeassistant.components.owntracks.device_tracker.get_cipher', @patch('homeassistant.components.owntracks.device_tracker.get_cipher',
mock_cipher) mock_cipher)
async def test_encrypted_payload_topic_key(hass, config_context): async def test_encrypted_payload_topic_key(hass, setup_comp):
"""Test encrypted payload with a topic key.""" """Test encrypted payload with a topic key."""
await setup_owntracks(hass, { await setup_owntracks(hass, {
CONF_SECRET: { CONF_SECRET: {
@ -1392,7 +1392,7 @@ async def test_encrypted_payload_topic_key(hass, config_context):
@patch('homeassistant.components.owntracks.device_tracker.get_cipher', @patch('homeassistant.components.owntracks.device_tracker.get_cipher',
mock_cipher) mock_cipher)
async def test_encrypted_payload_no_key(hass, config_context): async def test_encrypted_payload_no_key(hass, setup_comp):
"""Test encrypted payload with no key, .""" """Test encrypted payload with no key, ."""
assert hass.states.get(DEVICE_TRACKER_STATE) is None assert hass.states.get(DEVICE_TRACKER_STATE) is None
await setup_owntracks(hass, { await setup_owntracks(hass, {
@ -1405,7 +1405,7 @@ async def test_encrypted_payload_no_key(hass, config_context):
@patch('homeassistant.components.owntracks.device_tracker.get_cipher', @patch('homeassistant.components.owntracks.device_tracker.get_cipher',
mock_cipher) mock_cipher)
async def test_encrypted_payload_wrong_key(hass, config_context): async def test_encrypted_payload_wrong_key(hass, setup_comp):
"""Test encrypted payload with wrong key.""" """Test encrypted payload with wrong key."""
await setup_owntracks(hass, { await setup_owntracks(hass, {
CONF_SECRET: 'wrong key', CONF_SECRET: 'wrong key',
@ -1416,7 +1416,7 @@ async def test_encrypted_payload_wrong_key(hass, config_context):
@patch('homeassistant.components.owntracks.device_tracker.get_cipher', @patch('homeassistant.components.owntracks.device_tracker.get_cipher',
mock_cipher) mock_cipher)
async def test_encrypted_payload_wrong_topic_key(hass, config_context): async def test_encrypted_payload_wrong_topic_key(hass, setup_comp):
"""Test encrypted payload with wrong topic key.""" """Test encrypted payload with wrong topic key."""
await setup_owntracks(hass, { await setup_owntracks(hass, {
CONF_SECRET: { CONF_SECRET: {
@ -1429,7 +1429,7 @@ async def test_encrypted_payload_wrong_topic_key(hass, config_context):
@patch('homeassistant.components.owntracks.device_tracker.get_cipher', @patch('homeassistant.components.owntracks.device_tracker.get_cipher',
mock_cipher) mock_cipher)
async def test_encrypted_payload_no_topic_key(hass, config_context): async def test_encrypted_payload_no_topic_key(hass, setup_comp):
"""Test encrypted payload with no topic key.""" """Test encrypted payload with no topic key."""
await setup_owntracks(hass, { await setup_owntracks(hass, {
CONF_SECRET: { CONF_SECRET: {
@ -1439,9 +1439,10 @@ async def test_encrypted_payload_no_topic_key(hass, config_context):
assert hass.states.get(DEVICE_TRACKER_STATE) is None assert hass.states.get(DEVICE_TRACKER_STATE) is None
async def test_encrypted_payload_libsodium(hass, config_context): async def test_encrypted_payload_libsodium(hass, setup_comp):
"""Test sending encrypted message payload.""" """Test sending encrypted message payload."""
try: try:
# pylint: disable=unused-import
import libnacl # noqa: F401 import libnacl # noqa: F401
except (ImportError, OSError): except (ImportError, OSError):
pytest.skip("libnacl/libsodium is not installed") pytest.skip("libnacl/libsodium is not installed")
@ -1455,7 +1456,7 @@ async def test_encrypted_payload_libsodium(hass, config_context):
assert_location_latitude(hass, LOCATION_MESSAGE['lat']) assert_location_latitude(hass, LOCATION_MESSAGE['lat'])
async def test_customized_mqtt_topic(hass, config_context): async def test_customized_mqtt_topic(hass, setup_comp):
"""Test subscribing to a custom mqtt topic.""" """Test subscribing to a custom mqtt topic."""
await setup_owntracks(hass, { await setup_owntracks(hass, {
CONF_MQTT_TOPIC: 'mytracks/#', CONF_MQTT_TOPIC: 'mytracks/#',
@ -1467,7 +1468,7 @@ async def test_customized_mqtt_topic(hass, config_context):
assert_location_latitude(hass, LOCATION_MESSAGE['lat']) assert_location_latitude(hass, LOCATION_MESSAGE['lat'])
async def test_region_mapping(hass, config_context): async def test_region_mapping(hass, setup_comp):
"""Test region to zone mapping.""" """Test region to zone mapping."""
await setup_owntracks(hass, { await setup_owntracks(hass, {
CONF_REGION_MAPPING: { CONF_REGION_MAPPING: {