240 lines
7.3 KiB
Python
240 lines
7.3 KiB
Python
"""Webhook tests for mobile_app."""
|
|
|
|
import logging
|
|
|
|
import pytest
|
|
|
|
from homeassistant.components.mobile_app.const import CONF_SECRET
|
|
from homeassistant.components.zone import DOMAIN as ZONE_DOMAIN
|
|
from homeassistant.const import CONF_WEBHOOK_ID
|
|
from homeassistant.core import callback
|
|
from homeassistant.setup import async_setup_component
|
|
|
|
from .const import CALL_SERVICE, FIRE_EVENT, REGISTER_CLEARTEXT, RENDER_TEMPLATE, UPDATE
|
|
|
|
from tests.common import async_mock_service
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
async def test_webhook_handle_render_template(create_registrations, webhook_client):
|
|
"""Test that we render templates properly."""
|
|
resp = await webhook_client.post(
|
|
"/api/webhook/{}".format(create_registrations[1]["webhook_id"]),
|
|
json=RENDER_TEMPLATE,
|
|
)
|
|
|
|
assert resp.status == 200
|
|
|
|
json = await resp.json()
|
|
assert json == {"one": "Hello world"}
|
|
|
|
|
|
async def test_webhook_handle_call_services(hass, create_registrations, webhook_client):
|
|
"""Test that we call services properly."""
|
|
calls = async_mock_service(hass, "test", "mobile_app")
|
|
|
|
resp = await webhook_client.post(
|
|
"/api/webhook/{}".format(create_registrations[1]["webhook_id"]),
|
|
json=CALL_SERVICE,
|
|
)
|
|
|
|
assert resp.status == 200
|
|
|
|
assert len(calls) == 1
|
|
|
|
|
|
async def test_webhook_handle_fire_event(hass, create_registrations, webhook_client):
|
|
"""Test that we can fire events."""
|
|
events = []
|
|
|
|
@callback
|
|
def store_event(event):
|
|
"""Helepr to store events."""
|
|
events.append(event)
|
|
|
|
hass.bus.async_listen("test_event", store_event)
|
|
|
|
resp = await webhook_client.post(
|
|
"/api/webhook/{}".format(create_registrations[1]["webhook_id"]), json=FIRE_EVENT
|
|
)
|
|
|
|
assert resp.status == 200
|
|
json = await resp.json()
|
|
assert json == {}
|
|
|
|
assert len(events) == 1
|
|
assert events[0].data["hello"] == "yo world"
|
|
|
|
|
|
async def test_webhook_update_registration(webhook_client, hass_client):
|
|
"""Test that a we can update an existing registration via webhook."""
|
|
authed_api_client = await hass_client()
|
|
register_resp = await authed_api_client.post(
|
|
"/api/mobile_app/registrations", json=REGISTER_CLEARTEXT
|
|
)
|
|
|
|
assert register_resp.status == 201
|
|
register_json = await register_resp.json()
|
|
|
|
webhook_id = register_json[CONF_WEBHOOK_ID]
|
|
|
|
update_container = {"type": "update_registration", "data": UPDATE}
|
|
|
|
update_resp = await webhook_client.post(
|
|
"/api/webhook/{}".format(webhook_id), json=update_container
|
|
)
|
|
|
|
assert update_resp.status == 200
|
|
update_json = await update_resp.json()
|
|
assert update_json["app_version"] == "2.0.0"
|
|
assert CONF_WEBHOOK_ID not in update_json
|
|
assert CONF_SECRET not in update_json
|
|
|
|
|
|
async def test_webhook_handle_get_zones(hass, create_registrations, webhook_client):
|
|
"""Test that we can get zones properly."""
|
|
await async_setup_component(
|
|
hass,
|
|
ZONE_DOMAIN,
|
|
{
|
|
ZONE_DOMAIN: {
|
|
"name": "test",
|
|
"latitude": 32.880837,
|
|
"longitude": -117.237561,
|
|
"radius": 250,
|
|
}
|
|
},
|
|
)
|
|
|
|
resp = await webhook_client.post(
|
|
"/api/webhook/{}".format(create_registrations[1]["webhook_id"]),
|
|
json={"type": "get_zones"},
|
|
)
|
|
|
|
assert resp.status == 200
|
|
|
|
json = await resp.json()
|
|
assert len(json) == 1
|
|
assert json[0]["entity_id"] == "zone.home"
|
|
|
|
|
|
async def test_webhook_handle_get_config(hass, create_registrations, webhook_client):
|
|
"""Test that we can get config properly."""
|
|
resp = await webhook_client.post(
|
|
"/api/webhook/{}".format(create_registrations[1]["webhook_id"]),
|
|
json={"type": "get_config"},
|
|
)
|
|
|
|
assert resp.status == 200
|
|
|
|
json = await resp.json()
|
|
if "components" in json:
|
|
json["components"] = set(json["components"])
|
|
if "whitelist_external_dirs" in json:
|
|
json["whitelist_external_dirs"] = set(json["whitelist_external_dirs"])
|
|
|
|
hass_config = hass.config.as_dict()
|
|
|
|
expected_dict = {
|
|
"latitude": hass_config["latitude"],
|
|
"longitude": hass_config["longitude"],
|
|
"elevation": hass_config["elevation"],
|
|
"unit_system": hass_config["unit_system"],
|
|
"location_name": hass_config["location_name"],
|
|
"time_zone": hass_config["time_zone"],
|
|
"components": hass_config["components"],
|
|
"version": hass_config["version"],
|
|
"theme_color": "#03A9F4", # Default frontend theme color
|
|
}
|
|
|
|
assert expected_dict == json
|
|
|
|
|
|
async def test_webhook_returns_error_incorrect_json(
|
|
webhook_client, create_registrations, caplog
|
|
):
|
|
"""Test that an error is returned when JSON is invalid."""
|
|
resp = await webhook_client.post(
|
|
"/api/webhook/{}".format(create_registrations[1]["webhook_id"]), data="not json"
|
|
)
|
|
|
|
assert resp.status == 400
|
|
json = await resp.json()
|
|
assert json == {}
|
|
assert "invalid JSON" in caplog.text
|
|
|
|
|
|
async def test_webhook_handle_decryption(webhook_client, create_registrations):
|
|
"""Test that we can encrypt/decrypt properly."""
|
|
try:
|
|
from nacl.secret import SecretBox
|
|
from nacl.encoding import Base64Encoder
|
|
except (ImportError, OSError):
|
|
pytest.skip("libnacl/libsodium is not installed")
|
|
return
|
|
|
|
import json
|
|
|
|
keylen = SecretBox.KEY_SIZE
|
|
key = create_registrations[0]["secret"].encode("utf-8")
|
|
key = key[:keylen]
|
|
key = key.ljust(keylen, b"\0")
|
|
|
|
payload = json.dumps(RENDER_TEMPLATE["data"]).encode("utf-8")
|
|
|
|
data = SecretBox(key).encrypt(payload, encoder=Base64Encoder).decode("utf-8")
|
|
|
|
container = {"type": "render_template", "encrypted": True, "encrypted_data": data}
|
|
|
|
resp = await webhook_client.post(
|
|
"/api/webhook/{}".format(create_registrations[0]["webhook_id"]), json=container
|
|
)
|
|
|
|
assert resp.status == 200
|
|
|
|
webhook_json = await resp.json()
|
|
assert "encrypted_data" in webhook_json
|
|
|
|
decrypted_data = SecretBox(key).decrypt(
|
|
webhook_json["encrypted_data"], encoder=Base64Encoder
|
|
)
|
|
decrypted_data = decrypted_data.decode("utf-8")
|
|
|
|
assert json.loads(decrypted_data) == {"one": "Hello world"}
|
|
|
|
|
|
async def test_webhook_requires_encryption(webhook_client, create_registrations):
|
|
"""Test that encrypted registrations only accept encrypted data."""
|
|
resp = await webhook_client.post(
|
|
"/api/webhook/{}".format(create_registrations[0]["webhook_id"]),
|
|
json=RENDER_TEMPLATE,
|
|
)
|
|
|
|
assert resp.status == 400
|
|
|
|
webhook_json = await resp.json()
|
|
assert "error" in webhook_json
|
|
assert webhook_json["success"] is False
|
|
assert webhook_json["error"]["code"] == "encryption_required"
|
|
|
|
|
|
async def test_webhook_update_location(hass, webhook_client, create_registrations):
|
|
"""Test that encrypted registrations only accept encrypted data."""
|
|
resp = await webhook_client.post(
|
|
"/api/webhook/{}".format(create_registrations[1]["webhook_id"]),
|
|
json={
|
|
"type": "update_location",
|
|
"data": {"gps": [1, 2], "gps_accuracy": 10, "altitude": -10},
|
|
},
|
|
)
|
|
|
|
assert resp.status == 200
|
|
|
|
state = hass.states.get("device_tracker.test_1_2")
|
|
assert state is not None
|
|
assert state.attributes["latitude"] == 1.0
|
|
assert state.attributes["longitude"] == 2.0
|
|
assert state.attributes["gps_accuracy"] == 10
|
|
assert state.attributes["altitude"] == -10
|