a5f144cb7c
* Add action and callback support to html5 (#2855). Remove registrations from the callback view since we always get the latest anyway. We dont put an audience in the claims so we will never hit this error. Bring tests back up to where they were before callbacks. Only import jwt where necessary Fix bracket spacing errors Fix JWT decode check for loop Remove stale comment. Add tests for the callback system. Shorten line Disable pylint broad-except and change e to jwt_decode_error. Verify expiration Remove duplicate jwt.exceptions.DecodeError Catch no keys matched and return False * Switch to using registrations for callbackview instead of json_path * Only check for URL and such if the data object actually exists * raise instead of return * cleanup decode_jwt * Clean up JWT errors * Correctly set status_code to 401 * Improve JWT by adding target to claims and attempting to check the given target for a decode match first, as well as pass the target through in the event payload. * Add tag support and fix formatting issues * Pass through any keys that dont apply to the payload into the notification.data dictionary * Remove stale print * Pass back the data dictionary if it exists * Actually put the default url even if a notify payload dictionary doesnt exist * pylint, flake8 * Add subscription validation * Add validation for the callback event payload and use constants where possible * Use HTTP_UNAUTHORIZED instead of 401 * Change callback dictionary to dict instead of cv.match_all * Fix up tests and make subscription required * Whoops, that test was supposed to fail * Use the result of CALLBACK_EVENT_PAYLOAD_SCHEMA as event_payload * Add a test for html5 callback decode_jwt where the device has been renamed since notification has been sent. * Remove the loop through logic, assume that target is always in JWT * Always return something instead of possibly None. * Update frontend
240 lines
8.2 KiB
Python
240 lines
8.2 KiB
Python
"""Test HTML5 notify platform."""
|
|
import json
|
|
import tempfile
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from werkzeug.test import EnvironBuilder
|
|
|
|
from homeassistant.components.http import request_class
|
|
from homeassistant.components.notify import html5
|
|
|
|
|
|
class TestHtml5Notify(object):
|
|
"""Tests for HTML5 notify platform."""
|
|
|
|
def test_get_service_with_no_json(self):
|
|
"""Test empty json file."""
|
|
hass = MagicMock()
|
|
|
|
with tempfile.NamedTemporaryFile() as fp:
|
|
hass.config.path.return_value = fp.name
|
|
service = html5.get_service(hass, {})
|
|
|
|
assert service is not None
|
|
|
|
def test_get_service_with_bad_json(self):
|
|
"""Test ."""
|
|
hass = MagicMock()
|
|
|
|
with tempfile.NamedTemporaryFile() as fp:
|
|
fp.write('I am not JSON'.encode('utf-8'))
|
|
fp.flush()
|
|
hass.config.path.return_value = fp.name
|
|
service = html5.get_service(hass, {})
|
|
|
|
assert service is None
|
|
|
|
@patch('pywebpush.WebPusher')
|
|
def test_sending_message(self, mock_wp):
|
|
"""Test sending message."""
|
|
hass = MagicMock()
|
|
|
|
data = {
|
|
'device': {
|
|
'browser': 'chrome',
|
|
'subscription': {
|
|
'endpoint': 'https://google.com',
|
|
'keys': {'auth': 'auth', 'p256dh': 'p256dh'}
|
|
},
|
|
}
|
|
}
|
|
|
|
with tempfile.NamedTemporaryFile() as fp:
|
|
fp.write(json.dumps(data).encode('utf-8'))
|
|
fp.flush()
|
|
hass.config.path.return_value = fp.name
|
|
service = html5.get_service(hass, {'gcm_sender_id': '100'})
|
|
|
|
assert service is not None
|
|
|
|
service.send_message('Hello', target=['device', 'non_existing'],
|
|
data={'icon': 'beer.png'})
|
|
|
|
assert len(mock_wp.mock_calls) == 2
|
|
|
|
# WebPusher constructor
|
|
assert mock_wp.mock_calls[0][1][0] == {'endpoint':
|
|
'https://google.com',
|
|
'keys': {'auth': 'auth',
|
|
'p256dh': 'p256dh'}}
|
|
|
|
# Call to send
|
|
payload = json.loads(mock_wp.mock_calls[1][1][0])
|
|
|
|
assert payload['body'] == 'Hello'
|
|
assert payload['icon'] == 'beer.png'
|
|
|
|
def test_registering_new_device_view(self):
|
|
"""Test that the HTML view works."""
|
|
hass = MagicMock()
|
|
|
|
with tempfile.NamedTemporaryFile() as fp:
|
|
hass.config.path.return_value = fp.name
|
|
fp.close()
|
|
service = html5.get_service(hass, {})
|
|
|
|
assert service is not None
|
|
|
|
# assert hass.called
|
|
assert len(hass.mock_calls) == 3
|
|
|
|
view = hass.mock_calls[1][1][0]
|
|
assert view.json_path == fp.name
|
|
assert view.registrations == {}
|
|
|
|
builder = EnvironBuilder(method='POST', data=json.dumps({
|
|
'browser': 'chrome',
|
|
'subscription': {'endpoint': 'https://google.com',
|
|
'keys': {'auth': 'auth',
|
|
'p256dh': 'p256dh'}},
|
|
}))
|
|
Request = request_class()
|
|
resp = view.post(Request(builder.get_environ()))
|
|
|
|
expected = {
|
|
'unnamed device': {
|
|
'browser': 'chrome',
|
|
'subscription': {'endpoint': 'https://google.com',
|
|
'keys': {'auth': 'auth',
|
|
'p256dh': 'p256dh'}},
|
|
},
|
|
}
|
|
|
|
assert resp.status_code == 200, resp.response
|
|
assert view.registrations == expected
|
|
with open(fp.name) as fpp:
|
|
assert json.load(fpp) == expected
|
|
|
|
def test_registering_new_device_validation(self):
|
|
"""Test various errors when registering a new device."""
|
|
hass = MagicMock()
|
|
|
|
with tempfile.NamedTemporaryFile() as fp:
|
|
hass.config.path.return_value = fp.name
|
|
service = html5.get_service(hass, {})
|
|
|
|
assert service is not None
|
|
|
|
# assert hass.called
|
|
assert len(hass.mock_calls) == 3
|
|
|
|
view = hass.mock_calls[1][1][0]
|
|
|
|
Request = request_class()
|
|
|
|
builder = EnvironBuilder(method='POST', data=json.dumps({
|
|
'browser': 'invalid browser',
|
|
'subscription': 'sub info',
|
|
}))
|
|
resp = view.post(Request(builder.get_environ()))
|
|
assert resp.status_code == 400, resp.response
|
|
|
|
builder = EnvironBuilder(method='POST', data=json.dumps({
|
|
'browser': 'chrome',
|
|
}))
|
|
resp = view.post(Request(builder.get_environ()))
|
|
assert resp.status_code == 400, resp.response
|
|
|
|
builder = EnvironBuilder(method='POST', data=json.dumps({
|
|
'browser': 'chrome',
|
|
'subscription': 'sub info',
|
|
}))
|
|
with patch('homeassistant.components.notify.html5._save_config',
|
|
return_value=False):
|
|
resp = view.post(Request(builder.get_environ()))
|
|
assert resp.status_code == 400, resp.response
|
|
|
|
def test_callback_view_no_jwt(self):
|
|
"""Test that the notification callback view works without JWT."""
|
|
hass = MagicMock()
|
|
|
|
with tempfile.NamedTemporaryFile() as fp:
|
|
hass.config.path.return_value = fp.name
|
|
fp.close()
|
|
service = html5.get_service(hass, {})
|
|
|
|
assert service is not None
|
|
|
|
# assert hass.called
|
|
assert len(hass.mock_calls) == 3
|
|
|
|
view = hass.mock_calls[2][1][0]
|
|
|
|
builder = EnvironBuilder(method='POST', data=json.dumps({
|
|
'type': 'push',
|
|
'tag': '3bc28d69-0921-41f1-ac6a-7a627ba0aa72'
|
|
}))
|
|
Request = request_class()
|
|
resp = view.post(Request(builder.get_environ()))
|
|
|
|
assert resp.status_code == 401, resp.response
|
|
|
|
@patch('pywebpush.WebPusher')
|
|
def test_callback_view_with_jwt(self, mock_wp):
|
|
"""Test that the notification callback view works with JWT."""
|
|
hass = MagicMock()
|
|
|
|
data = {
|
|
'device': {
|
|
'browser': 'chrome',
|
|
'subscription': {
|
|
'endpoint': 'https://google.com',
|
|
'keys': {'auth': 'auth', 'p256dh': 'p256dh'}
|
|
},
|
|
}
|
|
}
|
|
|
|
with tempfile.NamedTemporaryFile() as fp:
|
|
fp.write(json.dumps(data).encode('utf-8'))
|
|
fp.flush()
|
|
hass.config.path.return_value = fp.name
|
|
service = html5.get_service(hass, {'gcm_sender_id': '100'})
|
|
|
|
assert service is not None
|
|
|
|
# assert hass.called
|
|
assert len(hass.mock_calls) == 3
|
|
|
|
service.send_message('Hello', target=['device'],
|
|
data={'icon': 'beer.png'})
|
|
|
|
assert len(mock_wp.mock_calls) == 2
|
|
|
|
# WebPusher constructor
|
|
assert mock_wp.mock_calls[0][1][0] == {'endpoint':
|
|
'https://google.com',
|
|
'keys': {'auth': 'auth',
|
|
'p256dh':
|
|
'p256dh'}}
|
|
|
|
# Call to send
|
|
push_payload = json.loads(mock_wp.mock_calls[1][1][0])
|
|
|
|
assert push_payload['body'] == 'Hello'
|
|
assert push_payload['icon'] == 'beer.png'
|
|
|
|
view = hass.mock_calls[2][1][0]
|
|
view.registrations = data
|
|
|
|
bearer_token = "Bearer {}".format(push_payload['data']['jwt'])
|
|
|
|
builder = EnvironBuilder(method='POST', data=json.dumps({
|
|
'type': 'push',
|
|
}), headers={'Authorization': bearer_token})
|
|
Request = request_class()
|
|
resp = view.post(Request(builder.get_environ()))
|
|
|
|
assert resp.status_code == 200, resp.response
|
|
returned = resp.response[0].decode('utf-8')
|
|
expected = '{"event": "push", "status": "ok"}'
|
|
assert json.loads(returned) == json.loads(expected)
|