mirror of
https://github.com/home-assistant/core.git
synced 2025-07-23 05:07:41 +00:00
Move imports in html5 component (#27473)
* Move imports in html5 component * Fix tests 1 * Fix tests 2
This commit is contained in:
parent
e992cfb45c
commit
7637ceb880
@ -9,6 +9,9 @@ import time
|
|||||||
import uuid
|
import uuid
|
||||||
|
|
||||||
from aiohttp.hdrs import AUTHORIZATION
|
from aiohttp.hdrs import AUTHORIZATION
|
||||||
|
import jwt
|
||||||
|
from pywebpush import WebPusher
|
||||||
|
from py_vapid import Vapid
|
||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
from voluptuous.humanize import humanize_error
|
from voluptuous.humanize import humanize_error
|
||||||
|
|
||||||
@ -311,7 +314,6 @@ class HTML5PushCallbackView(HomeAssistantView):
|
|||||||
|
|
||||||
def decode_jwt(self, token):
|
def decode_jwt(self, token):
|
||||||
"""Find the registration that signed this JWT and return it."""
|
"""Find the registration that signed this JWT and return it."""
|
||||||
import jwt
|
|
||||||
|
|
||||||
# 1. Check claims w/o verifying to see if a target is in there.
|
# 1. Check claims w/o verifying to see if a target is in there.
|
||||||
# 2. If target in claims, attempt to verify against the given name.
|
# 2. If target in claims, attempt to verify against the given name.
|
||||||
@ -335,7 +337,6 @@ class HTML5PushCallbackView(HomeAssistantView):
|
|||||||
# https://auth0.com/docs/quickstart/backend/python
|
# https://auth0.com/docs/quickstart/backend/python
|
||||||
def check_authorization_header(self, request):
|
def check_authorization_header(self, request):
|
||||||
"""Check the authorization header."""
|
"""Check the authorization header."""
|
||||||
import jwt
|
|
||||||
|
|
||||||
auth = request.headers.get(AUTHORIZATION, None)
|
auth = request.headers.get(AUTHORIZATION, None)
|
||||||
if not auth:
|
if not auth:
|
||||||
@ -491,7 +492,6 @@ class HTML5NotificationService(BaseNotificationService):
|
|||||||
|
|
||||||
def _push_message(self, payload, **kwargs):
|
def _push_message(self, payload, **kwargs):
|
||||||
"""Send the message."""
|
"""Send the message."""
|
||||||
from pywebpush import WebPusher
|
|
||||||
|
|
||||||
timestamp = int(time.time())
|
timestamp = int(time.time())
|
||||||
ttl = int(kwargs.get(ATTR_TTL, DEFAULT_TTL))
|
ttl = int(kwargs.get(ATTR_TTL, DEFAULT_TTL))
|
||||||
@ -550,7 +550,6 @@ class HTML5NotificationService(BaseNotificationService):
|
|||||||
|
|
||||||
def add_jwt(timestamp, target, tag, jwt_secret):
|
def add_jwt(timestamp, target, tag, jwt_secret):
|
||||||
"""Create JWT json to put into payload."""
|
"""Create JWT json to put into payload."""
|
||||||
import jwt
|
|
||||||
|
|
||||||
jwt_exp = datetime.fromtimestamp(timestamp) + timedelta(days=JWT_VALID_DAYS)
|
jwt_exp = datetime.fromtimestamp(timestamp) + timedelta(days=JWT_VALID_DAYS)
|
||||||
jwt_claims = {
|
jwt_claims = {
|
||||||
@ -565,7 +564,6 @@ def add_jwt(timestamp, target, tag, jwt_secret):
|
|||||||
|
|
||||||
def create_vapid_headers(vapid_email, subscription_info, vapid_private_key):
|
def create_vapid_headers(vapid_email, subscription_info, vapid_private_key):
|
||||||
"""Create encrypted headers to send to WebPusher."""
|
"""Create encrypted headers to send to WebPusher."""
|
||||||
from py_vapid import Vapid
|
|
||||||
|
|
||||||
if vapid_email and vapid_private_key and ATTR_ENDPOINT in subscription_info:
|
if vapid_email and vapid_private_key and ATTR_ENDPOINT in subscription_info:
|
||||||
url = urlparse(subscription_info.get(ATTR_ENDPOINT))
|
url = urlparse(subscription_info.get(ATTR_ENDPOINT))
|
||||||
|
@ -87,7 +87,7 @@ class TestHtml5Notify:
|
|||||||
|
|
||||||
assert service is not None
|
assert service is not None
|
||||||
|
|
||||||
@patch("pywebpush.WebPusher")
|
@patch("homeassistant.components.html5.notify.WebPusher")
|
||||||
def test_dismissing_message(self, mock_wp):
|
def test_dismissing_message(self, mock_wp):
|
||||||
"""Test dismissing message."""
|
"""Test dismissing message."""
|
||||||
hass = MagicMock()
|
hass = MagicMock()
|
||||||
@ -115,7 +115,7 @@ class TestHtml5Notify:
|
|||||||
assert payload["dismiss"] is True
|
assert payload["dismiss"] is True
|
||||||
assert payload["tag"] == "test"
|
assert payload["tag"] == "test"
|
||||||
|
|
||||||
@patch("pywebpush.WebPusher")
|
@patch("homeassistant.components.html5.notify.WebPusher")
|
||||||
def test_sending_message(self, mock_wp):
|
def test_sending_message(self, mock_wp):
|
||||||
"""Test sending message."""
|
"""Test sending message."""
|
||||||
hass = MagicMock()
|
hass = MagicMock()
|
||||||
@ -145,7 +145,7 @@ class TestHtml5Notify:
|
|||||||
assert payload["body"] == "Hello"
|
assert payload["body"] == "Hello"
|
||||||
assert payload["icon"] == "beer.png"
|
assert payload["icon"] == "beer.png"
|
||||||
|
|
||||||
@patch("pywebpush.WebPusher")
|
@patch("homeassistant.components.html5.notify.WebPusher")
|
||||||
def test_gcm_key_include(self, mock_wp):
|
def test_gcm_key_include(self, mock_wp):
|
||||||
"""Test if the gcm_key is only included for GCM endpoints."""
|
"""Test if the gcm_key is only included for GCM endpoints."""
|
||||||
hass = MagicMock()
|
hass = MagicMock()
|
||||||
@ -176,7 +176,7 @@ class TestHtml5Notify:
|
|||||||
assert mock_wp.mock_calls[1][2]["gcm_key"] is not None
|
assert mock_wp.mock_calls[1][2]["gcm_key"] is not None
|
||||||
assert mock_wp.mock_calls[4][2]["gcm_key"] is None
|
assert mock_wp.mock_calls[4][2]["gcm_key"] is None
|
||||||
|
|
||||||
@patch("pywebpush.WebPusher")
|
@patch("homeassistant.components.html5.notify.WebPusher")
|
||||||
def test_fcm_key_include(self, mock_wp):
|
def test_fcm_key_include(self, mock_wp):
|
||||||
"""Test if the FCM header is included."""
|
"""Test if the FCM header is included."""
|
||||||
hass = MagicMock()
|
hass = MagicMock()
|
||||||
@ -201,7 +201,7 @@ class TestHtml5Notify:
|
|||||||
# Get the keys passed to the WebPusher's send method
|
# Get the keys passed to the WebPusher's send method
|
||||||
assert mock_wp.mock_calls[1][2]["headers"]["Authorization"] is not None
|
assert mock_wp.mock_calls[1][2]["headers"]["Authorization"] is not None
|
||||||
|
|
||||||
@patch("pywebpush.WebPusher")
|
@patch("homeassistant.components.html5.notify.WebPusher")
|
||||||
def test_fcm_send_with_unknown_priority(self, mock_wp):
|
def test_fcm_send_with_unknown_priority(self, mock_wp):
|
||||||
"""Test if the gcm_key is only included for GCM endpoints."""
|
"""Test if the gcm_key is only included for GCM endpoints."""
|
||||||
hass = MagicMock()
|
hass = MagicMock()
|
||||||
@ -226,7 +226,7 @@ class TestHtml5Notify:
|
|||||||
# Get the keys passed to the WebPusher's send method
|
# Get the keys passed to the WebPusher's send method
|
||||||
assert mock_wp.mock_calls[1][2]["headers"]["priority"] == "normal"
|
assert mock_wp.mock_calls[1][2]["headers"]["priority"] == "normal"
|
||||||
|
|
||||||
@patch("pywebpush.WebPusher")
|
@patch("homeassistant.components.html5.notify.WebPusher")
|
||||||
def test_fcm_no_targets(self, mock_wp):
|
def test_fcm_no_targets(self, mock_wp):
|
||||||
"""Test if the gcm_key is only included for GCM endpoints."""
|
"""Test if the gcm_key is only included for GCM endpoints."""
|
||||||
hass = MagicMock()
|
hass = MagicMock()
|
||||||
@ -251,7 +251,7 @@ class TestHtml5Notify:
|
|||||||
# Get the keys passed to the WebPusher's send method
|
# Get the keys passed to the WebPusher's send method
|
||||||
assert mock_wp.mock_calls[1][2]["headers"]["priority"] == "normal"
|
assert mock_wp.mock_calls[1][2]["headers"]["priority"] == "normal"
|
||||||
|
|
||||||
@patch("pywebpush.WebPusher")
|
@patch("homeassistant.components.html5.notify.WebPusher")
|
||||||
def test_fcm_additional_data(self, mock_wp):
|
def test_fcm_additional_data(self, mock_wp):
|
||||||
"""Test if the gcm_key is only included for GCM endpoints."""
|
"""Test if the gcm_key is only included for GCM endpoints."""
|
||||||
hass = MagicMock()
|
hass = MagicMock()
|
||||||
@ -475,7 +475,7 @@ async def test_callback_view_with_jwt(hass, hass_client):
|
|||||||
registrations = {"device": SUBSCRIPTION_1}
|
registrations = {"device": SUBSCRIPTION_1}
|
||||||
client = await mock_client(hass, hass_client, registrations)
|
client = await mock_client(hass, hass_client, registrations)
|
||||||
|
|
||||||
with patch("pywebpush.WebPusher") as mock_wp:
|
with patch("homeassistant.components.html5.notify.WebPusher") as mock_wp:
|
||||||
await hass.services.async_call(
|
await hass.services.async_call(
|
||||||
"notify",
|
"notify",
|
||||||
"notify",
|
"notify",
|
||||||
@ -511,7 +511,7 @@ async def test_send_fcm_without_targets(hass, hass_client):
|
|||||||
"""Test that the notification is send with FCM without targets."""
|
"""Test that the notification is send with FCM without targets."""
|
||||||
registrations = {"device": SUBSCRIPTION_5}
|
registrations = {"device": SUBSCRIPTION_5}
|
||||||
await mock_client(hass, hass_client, registrations)
|
await mock_client(hass, hass_client, registrations)
|
||||||
with patch("pywebpush.WebPusher") as mock_wp:
|
with patch("homeassistant.components.html5.notify.WebPusher") as mock_wp:
|
||||||
await hass.services.async_call(
|
await hass.services.async_call(
|
||||||
"notify",
|
"notify",
|
||||||
"notify",
|
"notify",
|
||||||
|
Loading…
x
Reference in New Issue
Block a user