Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-24 12:59:34 +00:00
Code Activity

Switch locative to use the webhook component

Browse Source
This commit is contained in:
Rohan Kapoor
2019-01-12 19:18:33 -08:00
parent 04636e9ba7
commit 4e020b90e1
5 changed files with 158 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,11 +1,12 @@
"""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
@@ -36,8 +37,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 +66,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 +126,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 +137,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 +148,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 +158,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 +169,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 +177,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 +190,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 +201,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 +213,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 +222,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 +235,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