diff --git a/homeassistant/components/html5/const.py b/homeassistant/components/html5/const.py
index dd447b0e4c1b..08f8bbe266d8 100644
--- a/homeassistant/components/html5/const.py
+++ b/homeassistant/components/html5/const.py
@@ -25,3 +25,5 @@ ATTR_TIMESTAMP = "timestamp"
ATTR_TTL = "ttl"
ATTR_URGENCY = "urgency"
ATTR_VIBRATE = "vibrate"
+ATTR_SUBSCRIPTION = "subscription"
+ATTR_ENDPOINT = "endpoint"
diff --git a/homeassistant/components/html5/http.py b/homeassistant/components/html5/http.py
new file mode 100644
index 000000000000..e24a1b9213e9
--- /dev/null
+++ b/homeassistant/components/html5/http.py
@@ -0,0 +1,302 @@
+"""HTTP views for the HTML5 integration."""
+
+from contextlib import suppress
+from http import HTTPStatus
+import logging
+from typing import Any, cast
+import warnings
+
+from aiohttp import web
+from aiohttp.hdrs import AUTHORIZATION
+import jwt
+from jwt.warnings import InsecureKeyLengthWarning
+import voluptuous as vol
+from voluptuous.humanize import humanize_error
+
+from homeassistant.components.http import KEY_HASS, HomeAssistantView
+from homeassistant.components.notify import ATTR_DATA, ATTR_TARGET
+from homeassistant.const import ATTR_NAME
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import config_validation as cv
+from homeassistant.helpers.dispatcher import async_dispatcher_send
+from homeassistant.helpers.json import save_json
+from homeassistant.util import ensure_unique_string
+
+from .const import ATTR_ACTION, ATTR_ENDPOINT, ATTR_SUBSCRIPTION, ATTR_TAG, DOMAIN
+from .entity import Registration
+from .issue import deprecated_event_bus
+
+_LOGGER = logging.getLogger(__name__)
+
+
+ATTR_TYPE = "type"
+ATTR_BROWSER = "browser"
+ATTR_KEYS = "keys"
+ATTR_AUTH = "auth"
+ATTR_P256DH = "p256dh"
+ATTR_EXPIRATIONTIME = "expirationTime"
+NOTIFY_CALLBACK_EVENT = "html5_notification"
+
+
+KEYS_SCHEMA = vol.All(
+ dict,
+ vol.Schema(
+ {
+ vol.Required(ATTR_AUTH): cv.string,
+ vol.Required(ATTR_P256DH): cv.string,
+ }
+ ),
+)
+
+SUBSCRIPTION_SCHEMA = vol.All(
+ dict,
+ vol.Schema(
+ {
+ vol.Required(ATTR_ENDPOINT): vol.Url(),
+ vol.Required(ATTR_KEYS): KEYS_SCHEMA,
+ vol.Optional(ATTR_EXPIRATIONTIME): vol.Any(None, cv.positive_int),
+ }
+ ),
+)
+
+REGISTER_SCHEMA = vol.Schema(
+ {
+ vol.Required(ATTR_SUBSCRIPTION): SUBSCRIPTION_SCHEMA,
+ vol.Required(ATTR_BROWSER): vol.In(["chrome", "firefox"]),
+ vol.Optional(ATTR_NAME): cv.string,
+ }
+)
+
+CALLBACK_EVENT_PAYLOAD_SCHEMA = vol.Schema(
+ {
+ vol.Required(ATTR_TAG): cv.string,
+ vol.Required(ATTR_TYPE): vol.In(["received", "clicked", "closed"]),
+ vol.Required(ATTR_TARGET): cv.string,
+ vol.Optional(ATTR_ACTION): cv.string,
+ vol.Optional(ATTR_DATA): dict,
+ }
+)
+
+
+@callback
+def async_register_http_views(
+ hass: HomeAssistant, json_path: str, registrations: dict[str, Registration]
+) -> None:
+ """Register the http views."""
+
+ hass.http.register_view(HTML5PushRegistrationView(registrations, json_path))
+ hass.http.register_view(HTML5PushCallbackView(registrations))
+
+
+class HTML5PushRegistrationView(HomeAssistantView):
+ """Accepts push registrations from a browser."""
+
+ url = "/api/notify.html5"
+ name = "api:notify.html5"
+
+ def __init__(self, registrations: dict[str, Registration], json_path: str) -> None:
+ """Init HTML5PushRegistrationView."""
+ self.registrations = registrations
+ self.json_path = json_path
+
+ async def post(self, request: web.Request) -> web.Response:
+ """Accept the POST request for push registrations from a browser."""
+
+ try:
+ data: Registration = await request.json()
+ except ValueError:
+ return self.json_message("Invalid JSON", HTTPStatus.BAD_REQUEST)
+ try:
+ data = cast(Registration, REGISTER_SCHEMA(data))
+ except vol.Invalid as ex:
+ return self.json_message(humanize_error(data, ex), HTTPStatus.BAD_REQUEST)
+
+ devname = data.get(ATTR_NAME)
+ data.pop(ATTR_NAME, None)
+
+ name = self.find_registration_name(data, devname)
+ previous_registration = self.registrations.get(name)
+
+ self.registrations[name] = data
+ hass = request.app[KEY_HASS]
+
+ try:
+ await hass.async_add_executor_job(
+ save_json, self.json_path, self.registrations
+ )
+ except HomeAssistantError:
+ if previous_registration is not None:
+ self.registrations[name] = previous_registration
+ else:
+ self.registrations.pop(name)
+
+ return self.json_message(
+ "Error saving registration.", HTTPStatus.INTERNAL_SERVER_ERROR
+ )
+
+ return self.json_message("Push notification subscriber registered.")
+
+ def find_registration_name(
+ self,
+ data: Registration,
+ suggested: str | None = None,
+ ):
+ """Find a registration name matching data or generate a unique one."""
+ endpoint = data["subscription"]["endpoint"]
+ for key, registration in self.registrations.items():
+ subscription = registration["subscription"]
+ if subscription.get(ATTR_ENDPOINT) == endpoint:
+ return key
+ return ensure_unique_string(suggested or "unnamed device", self.registrations)
+
+ async def delete(self, request: web.Request):
+ """Delete a registration."""
+ try:
+ data: dict[str, Any] = await request.json()
+ except ValueError:
+ return self.json_message("Invalid JSON", HTTPStatus.BAD_REQUEST)
+
+ subscription: dict[str, Any] = data[ATTR_SUBSCRIPTION]
+
+ found = None
+
+ for key, registration in self.registrations.items():
+ if registration["subscription"] == subscription:
+ found = key
+ break
+
+ if not found:
+ # If not found, unregistering was already done. Return 200
+ return self.json_message("Registration not found.")
+
+ reg = self.registrations.pop(found)
+ hass = request.app[KEY_HASS]
+
+ try:
+ await hass.async_add_executor_job(
+ save_json, self.json_path, self.registrations
+ )
+ except HomeAssistantError:
+ self.registrations[found] = reg
+ return self.json_message(
+ "Error saving registration.", HTTPStatus.INTERNAL_SERVER_ERROR
+ )
+
+ return self.json_message("Push notification subscriber unregistered.")
+
+
+class HTML5PushCallbackView(HomeAssistantView):
+ """Accepts push registrations from a browser."""
+
+ requires_auth = False
+ url = "/api/notify.html5/callback"
+ name = "api:notify.html5/callback"
+
+ def __init__(self, registrations: dict[str, Registration]) -> None:
+ """Init HTML5PushCallbackView."""
+ self.registrations = registrations
+
+ def decode_jwt(self, token: str) -> web.Response | dict[str, Any]:
+ """Find the registration that signed this JWT and return it."""
+
+ # 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.
+ # 2a. If decode is successful, return the payload.
+ # 2b. If decode is unsuccessful, return a 401.
+
+ target_check: dict[str, Any] = jwt.decode(
+ token, algorithms=["ES256", "HS256"], options={"verify_signature": False}
+ )
+ if target_check.get(ATTR_TARGET) in self.registrations:
+ possible_target = self.registrations[target_check[ATTR_TARGET]]
+ key = possible_target["subscription"]["keys"]["auth"]
+ with (
+ suppress(jwt.exceptions.DecodeError, jwt.exceptions.InvalidKeyError),
+ warnings.catch_warnings(),
+ ):
+ warnings.simplefilter("ignore", InsecureKeyLengthWarning)
+ return jwt.decode(token, key, algorithms=["ES256", "HS256"])
+
+ return self.json_message(
+ "No target found in JWT", status_code=HTTPStatus.UNAUTHORIZED
+ )
+
+ # The following is based on code from Auth0
+ # https://auth0.com/docs/quickstart/backend/python
+ def check_authorization_header(
+ self, request: web.Request
+ ) -> web.Response | dict[str, Any]:
+ """Check the authorization header."""
+ if not (auth := request.headers.get(AUTHORIZATION)):
+ return self.json_message(
+ "Authorization header is expected", status_code=HTTPStatus.UNAUTHORIZED
+ )
+
+ parts = auth.split()
+
+ if parts[0].lower() != "bearer":
+ return self.json_message(
+ "Authorization header must start with Bearer",
+ status_code=HTTPStatus.UNAUTHORIZED,
+ )
+ if len(parts) != 2:
+ return self.json_message(
+ "Authorization header must be Bearer token",
+ status_code=HTTPStatus.UNAUTHORIZED,
+ )
+
+ token = parts[1]
+ try:
+ payload = self.decode_jwt(token)
+ except jwt.exceptions.InvalidTokenError:
+ return self.json_message(
+ "token is invalid", status_code=HTTPStatus.UNAUTHORIZED
+ )
+ return payload
+
+ async def post(self, request: web.Request) -> web.Response:
+ """Accept the POST request for push registrations event callback."""
+ auth_check = self.check_authorization_header(request)
+ if not isinstance(auth_check, dict):
+ return auth_check
+
+ try:
+ data: dict[str, str] = await request.json()
+ except ValueError:
+ return self.json_message("Invalid JSON", HTTPStatus.BAD_REQUEST)
+
+ event_payload: dict[str, Any] = {
+ ATTR_TAG: data.get(ATTR_TAG),
+ ATTR_TYPE: data[ATTR_TYPE],
+ ATTR_TARGET: auth_check[ATTR_TARGET],
+ }
+
+ if data.get(ATTR_ACTION) is not None:
+ event_payload[ATTR_ACTION] = data.get(ATTR_ACTION)
+
+ if data.get(ATTR_DATA) is not None:
+ event_payload[ATTR_DATA] = data.get(ATTR_DATA)
+
+ try:
+ event_payload = CALLBACK_EVENT_PAYLOAD_SCHEMA(event_payload)
+ except vol.Invalid as ex:
+ _LOGGER.warning(
+ "Callback event payload is not valid: %s",
+ humanize_error(event_payload, ex),
+ )
+
+ event_name = f"{NOTIFY_CALLBACK_EVENT}.{event_payload[ATTR_TYPE]}"
+ hass = request.app[KEY_HASS]
+ hass.bus.fire(event_name, event_payload)
+ async_dispatcher_send(
+ hass,
+ DOMAIN,
+ event_payload[ATTR_TARGET],
+ event_payload[ATTR_TYPE],
+ event_payload,
+ )
+
+ deprecated_event_bus(hass, event_name)
+
+ return self.json({"status": "ok", "event": event_payload[ATTR_TYPE]})
diff --git a/homeassistant/components/html5/notify.py b/homeassistant/components/html5/notify.py
index f56c4facb4cb..3b6148af46e7 100644
--- a/homeassistant/components/html5/notify.py
+++ b/homeassistant/components/html5/notify.py
@@ -11,17 +11,14 @@ from urllib.parse import urlparse
import uuid
import warnings
-from aiohttp import ClientError, ClientResponse, ClientSession, web
-from aiohttp.hdrs import AUTHORIZATION
+from aiohttp import ClientError, ClientResponse, ClientSession
import jwt
from jwt.warnings import InsecureKeyLengthWarning
from py_vapid import Vapid
from pywebpush import WebPusher, WebPushException, webpush_async
import voluptuous as vol
-from voluptuous.humanize import humanize_error
from homeassistant.components import websocket_api
-from homeassistant.components.http import KEY_HASS, HomeAssistantView
from homeassistant.components.notify import (
ATTR_DATA,
ATTR_TARGET,
@@ -33,20 +30,17 @@ from homeassistant.components.notify import (
)
from homeassistant.components.websocket_api import ActiveConnection
from homeassistant.config_entries import ConfigEntry
-from homeassistant.const import ATTR_NAME, URL_ROOT
+from homeassistant.const import URL_ROOT
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
-from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.json import save_json
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
-from homeassistant.util import ensure_unique_string
from homeassistant.util.json import load_json_object
from .const import (
- ATTR_ACTION,
ATTR_ACTIONS,
ATTR_REQUIRE_INTERACTION,
ATTR_TAG,
@@ -60,25 +54,12 @@ from .const import (
SERVICE_DISMISS,
)
from .entity import HTML5Entity, Registration
-from .issue import (
- deprecated_dismiss_action_call,
- deprecated_event_bus,
- deprecated_notify_action_call,
-)
+from .http import REGISTER_SCHEMA, async_register_http_views
+from .issue import deprecated_dismiss_action_call, deprecated_notify_action_call
_LOGGER = logging.getLogger(__name__)
-ATTR_SUBSCRIPTION = "subscription"
-ATTR_BROWSER = "browser"
-
-ATTR_ENDPOINT = "endpoint"
-ATTR_KEYS = "keys"
-ATTR_AUTH = "auth"
-ATTR_P256DH = "p256dh"
-ATTR_EXPIRATIONTIME = "expirationTime"
-
-ATTR_TYPE = "type"
ATTR_URL = "url"
ATTR_DISMISS = "dismiss"
ATTR_PRIORITY = "priority"
@@ -100,23 +81,6 @@ SCHEMA_WS_APPKEY = websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend(
JWT_VALID_DAYS = 7
VAPID_CLAIM_VALID_HOURS = 12
-KEYS_SCHEMA = vol.All(
- dict,
- vol.Schema(
- {vol.Required(ATTR_AUTH): cv.string, vol.Required(ATTR_P256DH): cv.string}
- ),
-)
-
-SUBSCRIPTION_SCHEMA = vol.All(
- dict,
- vol.Schema(
- {
- vol.Required(ATTR_ENDPOINT): vol.Url(),
- vol.Required(ATTR_KEYS): KEYS_SCHEMA,
- vol.Optional(ATTR_EXPIRATIONTIME): vol.Any(None, cv.positive_int),
- }
- ),
-)
DISMISS_SERVICE_SCHEMA = vol.Schema(
{
@@ -125,25 +89,6 @@ DISMISS_SERVICE_SCHEMA = vol.Schema(
}
)
-REGISTER_SCHEMA = vol.Schema(
- {
- vol.Required(ATTR_SUBSCRIPTION): SUBSCRIPTION_SCHEMA,
- vol.Required(ATTR_BROWSER): vol.In(["chrome", "firefox"]),
- vol.Optional(ATTR_NAME): cv.string,
- }
-)
-
-CALLBACK_EVENT_PAYLOAD_SCHEMA = vol.Schema(
- {
- vol.Required(ATTR_TAG): cv.string,
- vol.Required(ATTR_TYPE): vol.In(["received", "clicked", "closed"]),
- vol.Required(ATTR_TARGET): cv.string,
- vol.Optional(ATTR_ACTION): cv.string,
- vol.Optional(ATTR_DATA): dict,
- }
-)
-
-NOTIFY_CALLBACK_EVENT = "html5_notification"
# Badge and timestamp are Chrome specific (not in official spec)
HTML5_SHOWNOTIFICATION_PARAMETERS = (
@@ -192,8 +137,7 @@ async def async_get_service(
hass, WS_TYPE_APPKEY, websocket_appkey, SCHEMA_WS_APPKEY
)
- hass.http.register_view(HTML5PushRegistrationView(registrations, json_path))
- hass.http.register_view(HTML5PushCallbackView(registrations))
+ async_register_http_views(hass, json_path, registrations)
session = async_get_clientsession(hass)
return HTML5NotificationService(
@@ -208,220 +152,6 @@ def _load_config(filename: str) -> dict[str, Registration]:
return {}
-class HTML5PushRegistrationView(HomeAssistantView):
- """Accepts push registrations from a browser."""
-
- url = "/api/notify.html5"
- name = "api:notify.html5"
-
- def __init__(self, registrations: dict[str, Registration], json_path: str) -> None:
- """Init HTML5PushRegistrationView."""
- self.registrations = registrations
- self.json_path = json_path
-
- async def post(self, request: web.Request) -> web.Response:
- """Accept the POST request for push registrations from a browser."""
-
- try:
- data: Registration = await request.json()
- except ValueError:
- return self.json_message("Invalid JSON", HTTPStatus.BAD_REQUEST)
- try:
- data = cast(Registration, REGISTER_SCHEMA(data))
- except vol.Invalid as ex:
- return self.json_message(humanize_error(data, ex), HTTPStatus.BAD_REQUEST)
-
- devname = data.get(ATTR_NAME)
- data.pop(ATTR_NAME, None)
-
- name = self.find_registration_name(data, devname)
- previous_registration = self.registrations.get(name)
-
- self.registrations[name] = data
-
- try:
- hass = request.app[KEY_HASS]
-
- await hass.async_add_executor_job(
- save_json, self.json_path, self.registrations
- )
- return self.json_message("Push notification subscriber registered.")
- except HomeAssistantError:
- if previous_registration is not None:
- self.registrations[name] = previous_registration
- else:
- self.registrations.pop(name)
-
- return self.json_message(
- "Error saving registration.", HTTPStatus.INTERNAL_SERVER_ERROR
- )
-
- def find_registration_name(
- self,
- data: Registration,
- suggested: str | None = None,
- ):
- """Find a registration name matching data or generate a unique one."""
- endpoint = data["subscription"]["endpoint"]
- for key, registration in self.registrations.items():
- subscription = registration["subscription"]
- if subscription.get(ATTR_ENDPOINT) == endpoint:
- return key
- return ensure_unique_string(suggested or "unnamed device", self.registrations)
-
- async def delete(self, request: web.Request):
- """Delete a registration."""
- try:
- data: dict[str, Any] = await request.json()
- except ValueError:
- return self.json_message("Invalid JSON", HTTPStatus.BAD_REQUEST)
-
- subscription: dict[str, Any] = data[ATTR_SUBSCRIPTION]
-
- found = None
-
- for key, registration in self.registrations.items():
- if registration["subscription"] == subscription:
- found = key
- break
-
- if not found:
- # If not found, unregistering was already done. Return 200
- return self.json_message("Registration not found.")
-
- reg = self.registrations.pop(found)
-
- try:
- hass = request.app[KEY_HASS]
-
- await hass.async_add_executor_job(
- save_json, self.json_path, self.registrations
- )
- except HomeAssistantError:
- self.registrations[found] = reg
- return self.json_message(
- "Error saving registration.", HTTPStatus.INTERNAL_SERVER_ERROR
- )
-
- return self.json_message("Push notification subscriber unregistered.")
-
-
-class HTML5PushCallbackView(HomeAssistantView):
- """Accepts push registrations from a browser."""
-
- requires_auth = False
- url = "/api/notify.html5/callback"
- name = "api:notify.html5/callback"
-
- def __init__(self, registrations: dict[str, Registration]) -> None:
- """Init HTML5PushCallbackView."""
- self.registrations = registrations
-
- def decode_jwt(self, token: str) -> web.Response | dict[str, Any]:
- """Find the registration that signed this JWT and return it."""
-
- # 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.
- # 2a. If decode is successful, return the payload.
- # 2b. If decode is unsuccessful, return a 401.
-
- target_check: dict[str, Any] = jwt.decode(
- token, algorithms=["ES256", "HS256"], options={"verify_signature": False}
- )
- if target_check.get(ATTR_TARGET) in self.registrations:
- possible_target = self.registrations[target_check[ATTR_TARGET]]
- key = possible_target["subscription"]["keys"]["auth"]
- with (
- suppress(jwt.exceptions.DecodeError, jwt.exceptions.InvalidKeyError),
- warnings.catch_warnings(),
- ):
- warnings.simplefilter("ignore", InsecureKeyLengthWarning)
- return jwt.decode(token, key, algorithms=["ES256", "HS256"])
-
- return self.json_message(
- "No target found in JWT", status_code=HTTPStatus.UNAUTHORIZED
- )
-
- # The following is based on code from Auth0
- # https://auth0.com/docs/quickstart/backend/python
- def check_authorization_header(
- self, request: web.Request
- ) -> web.Response | dict[str, Any]:
- """Check the authorization header."""
- if not (auth := request.headers.get(AUTHORIZATION)):
- return self.json_message(
- "Authorization header is expected", status_code=HTTPStatus.UNAUTHORIZED
- )
-
- parts = auth.split()
-
- if parts[0].lower() != "bearer":
- return self.json_message(
- "Authorization header must start with Bearer",
- status_code=HTTPStatus.UNAUTHORIZED,
- )
- if len(parts) != 2:
- return self.json_message(
- "Authorization header must be Bearer token",
- status_code=HTTPStatus.UNAUTHORIZED,
- )
-
- token = parts[1]
- try:
- payload = self.decode_jwt(token)
- except jwt.exceptions.InvalidTokenError:
- return self.json_message(
- "token is invalid", status_code=HTTPStatus.UNAUTHORIZED
- )
- return payload
-
- async def post(self, request: web.Request) -> web.Response:
- """Accept the POST request for push registrations event callback."""
- auth_check = self.check_authorization_header(request)
- if not isinstance(auth_check, dict):
- return auth_check
-
- try:
- data: dict[str, str] = await request.json()
- except ValueError:
- return self.json_message("Invalid JSON", HTTPStatus.BAD_REQUEST)
-
- event_payload: dict[str, Any] = {
- ATTR_TAG: data.get(ATTR_TAG),
- ATTR_TYPE: data[ATTR_TYPE],
- ATTR_TARGET: auth_check[ATTR_TARGET],
- }
-
- if data.get(ATTR_ACTION) is not None:
- event_payload[ATTR_ACTION] = data.get(ATTR_ACTION)
-
- if data.get(ATTR_DATA) is not None:
- event_payload[ATTR_DATA] = data.get(ATTR_DATA)
-
- try:
- event_payload = CALLBACK_EVENT_PAYLOAD_SCHEMA(event_payload)
- except vol.Invalid as ex:
- _LOGGER.warning(
- "Callback event payload is not valid: %s",
- humanize_error(event_payload, ex),
- )
-
- event_name = f"{NOTIFY_CALLBACK_EVENT}.{event_payload[ATTR_TYPE]}"
- hass = request.app[KEY_HASS]
- hass.bus.fire(event_name, event_payload)
- async_dispatcher_send(
- hass,
- DOMAIN,
- event_payload[ATTR_TARGET],
- event_payload[ATTR_TYPE],
- event_payload,
- )
-
- deprecated_event_bus(hass, event_name)
-
- return self.json({"status": "ok", "event": event_payload[ATTR_TYPE]})
-
-
class HTML5NotificationService(BaseNotificationService):
"""Implement the notification service for HTML5."""
diff --git a/tests/components/html5/conftest.py b/tests/components/html5/conftest.py
index b818dbe6b7e3..5b79424a1dbb 100644
--- a/tests/components/html5/conftest.py
+++ b/tests/components/html5/conftest.py
@@ -91,6 +91,7 @@ def mock_jwt() -> Generator[MagicMock]:
with (
patch("homeassistant.components.html5.notify.jwt") as mock_client,
+ patch("homeassistant.components.html5.http.jwt", new=mock_client),
):
mock_client.encode.return_value = "JWT"
mock_client.decode.return_value = {"target": "device"}
@@ -123,3 +124,14 @@ def mock_vapid() -> Generator[MagicMock]:
"priority": "normal",
}
yield mock_client
+
+
+@pytest.fixture
+def mock_save() -> Generator[MagicMock]:
+ """Mock save_json."""
+
+ with (
+ patch("homeassistant.components.html5.http.save_json") as mock_client,
+ patch("homeassistant.components.html5.notify.save_json", new=mock_client),
+ ):
+ yield mock_client
diff --git a/tests/components/html5/test_event.py b/tests/components/html5/test_event.py
index 25e783df5e6b..a97689b64dc0 100644
--- a/tests/components/html5/test_event.py
+++ b/tests/components/html5/test_event.py
@@ -9,8 +9,8 @@ from aiohttp.hdrs import AUTHORIZATION
import pytest
from syrupy.assertion import SnapshotAssertion
-from homeassistant.components.html5.const import DOMAIN
-from homeassistant.components.html5.notify import ATTR_ACTION, ATTR_TAG, ATTR_TYPE
+from homeassistant.components.html5.const import ATTR_ACTION, ATTR_TAG, DOMAIN
+from homeassistant.components.html5.http import ATTR_TYPE
from homeassistant.components.notify import ATTR_DATA, ATTR_TARGET
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import STATE_UNKNOWN, Platform
diff --git a/tests/components/html5/test_notify.py b/tests/components/html5/test_notify.py
index 564aa3fc92f6..77320ab8c2eb 100644
--- a/tests/components/html5/test_notify.py
+++ b/tests/components/html5/test_notify.py
@@ -31,7 +31,8 @@ from homeassistant.components.html5.const import (
ATTR_VIBRATE,
SERVICE_DISMISS,
)
-from homeassistant.components.html5.notify import ATTR_ACTION, ATTR_DISMISS, DEFAULT_TTL
+from homeassistant.components.html5.http import ATTR_ACTION
+from homeassistant.components.html5.notify import ATTR_DISMISS, DEFAULT_TTL
from homeassistant.components.html5.services import SERVICE_DISMISS_MESSAGE
from homeassistant.components.notify import (
ATTR_DATA,
@@ -347,6 +348,7 @@ async def test_registering_new_device_view(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
+ mock_save: MagicMock,
) -> None:
"""Test that the HTML view works."""
await async_setup_component(hass, "http", {})
@@ -359,8 +361,7 @@ async def test_registering_new_device_view(
client = await hass_client()
- with patch("homeassistant.components.html5.notify.save_json") as mock_save:
- resp = await client.post(REGISTER_URL, data=json.dumps(SUBSCRIPTION_1))
+ resp = await client.post(REGISTER_URL, data=json.dumps(SUBSCRIPTION_1))
assert resp.status == HTTPStatus.OK
assert len(mock_save.mock_calls) == 1
@@ -372,6 +373,7 @@ async def test_registering_new_device_view_with_name(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
+ mock_save: MagicMock,
) -> None:
"""Test that the HTML view works with name attribute."""
await async_setup_component(hass, "http", {})
@@ -387,8 +389,7 @@ async def test_registering_new_device_view_with_name(
SUB_WITH_NAME = SUBSCRIPTION_1.copy()
SUB_WITH_NAME["name"] = "test device"
- with patch("homeassistant.components.html5.notify.save_json") as mock_save:
- resp = await client.post(REGISTER_URL, data=json.dumps(SUB_WITH_NAME))
+ resp = await client.post(REGISTER_URL, data=json.dumps(SUB_WITH_NAME))
assert resp.status == HTTPStatus.OK
assert len(mock_save.mock_calls) == 1
@@ -400,6 +401,7 @@ async def test_registering_new_device_expiration_view(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
+ mock_save: MagicMock,
) -> None:
"""Test that the HTML view works."""
await async_setup_component(hass, "http", {})
@@ -412,8 +414,7 @@ async def test_registering_new_device_expiration_view(
client = await hass_client()
- with patch("homeassistant.components.html5.notify.save_json") as mock_save:
- resp = await client.post(REGISTER_URL, data=json.dumps(SUBSCRIPTION_4))
+ resp = await client.post(REGISTER_URL, data=json.dumps(SUBSCRIPTION_4))
assert resp.status == HTTPStatus.OK
assert mock_save.mock_calls[0][1][1] == {"unnamed device": SUBSCRIPTION_4}
@@ -424,6 +425,7 @@ async def test_registering_new_device_fails_view(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
+ mock_save: MagicMock,
) -> None:
"""Test subs. are not altered when registering a new device fails."""
await async_setup_component(hass, "http", {})
@@ -435,11 +437,9 @@ async def test_registering_new_device_fails_view(
assert config_entry.state is ConfigEntryState.LOADED
client = await hass_client()
- with patch(
- "homeassistant.components.html5.notify.save_json",
- side_effect=HomeAssistantError(),
- ):
- resp = await client.post(REGISTER_URL, data=json.dumps(SUBSCRIPTION_4))
+ mock_save.side_effect = (HomeAssistantError(),)
+
+ resp = await client.post(REGISTER_URL, data=json.dumps(SUBSCRIPTION_4))
assert resp.status == HTTPStatus.INTERNAL_SERVER_ERROR
@@ -449,6 +449,7 @@ async def test_registering_existing_device_view(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
+ mock_save: MagicMock,
) -> None:
"""Test subscription is updated when registering existing device."""
await async_setup_component(hass, "http", {})
@@ -461,9 +462,8 @@ async def test_registering_existing_device_view(
client = await hass_client()
- with patch("homeassistant.components.html5.notify.save_json") as mock_save:
- await client.post(REGISTER_URL, data=json.dumps(SUBSCRIPTION_1))
- resp = await client.post(REGISTER_URL, data=json.dumps(SUBSCRIPTION_4))
+ await client.post(REGISTER_URL, data=json.dumps(SUBSCRIPTION_1))
+ resp = await client.post(REGISTER_URL, data=json.dumps(SUBSCRIPTION_4))
assert resp.status == HTTPStatus.OK
mock_save.assert_called_with(
@@ -476,6 +476,7 @@ async def test_registering_existing_device_view_with_name(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
+ mock_save: MagicMock,
) -> None:
"""Test subscription is updated when reg'ing existing device with name."""
await async_setup_component(hass, "http", {})
@@ -491,9 +492,8 @@ async def test_registering_existing_device_view_with_name(
SUB_WITH_NAME = SUBSCRIPTION_1.copy()
SUB_WITH_NAME["name"] = "test device"
- with patch("homeassistant.components.html5.notify.save_json") as mock_save:
- await client.post(REGISTER_URL, data=json.dumps(SUB_WITH_NAME))
- resp = await client.post(REGISTER_URL, data=json.dumps(SUBSCRIPTION_4))
+ await client.post(REGISTER_URL, data=json.dumps(SUB_WITH_NAME))
+ resp = await client.post(REGISTER_URL, data=json.dumps(SUBSCRIPTION_4))
assert resp.status == HTTPStatus.OK
@@ -507,6 +507,7 @@ async def test_registering_existing_device_fails_view(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
+ mock_save: MagicMock,
) -> None:
"""Test sub. is not updated when registering existing device fails."""
await async_setup_component(hass, "http", {})
@@ -519,10 +520,9 @@ async def test_registering_existing_device_fails_view(
client = await hass_client()
- with patch("homeassistant.components.html5.notify.save_json") as mock_save:
- await client.post(REGISTER_URL, data=json.dumps(SUBSCRIPTION_1))
- mock_save.side_effect = HomeAssistantError
- resp = await client.post(REGISTER_URL, data=json.dumps(SUBSCRIPTION_4))
+ await client.post(REGISTER_URL, data=json.dumps(SUBSCRIPTION_1))
+ mock_save.side_effect = HomeAssistantError
+ resp = await client.post(REGISTER_URL, data=json.dumps(SUBSCRIPTION_4))
assert resp.status == HTTPStatus.INTERNAL_SERVER_ERROR
@@ -532,6 +532,7 @@ async def test_registering_new_device_validation(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
+ mock_save: MagicMock,
) -> None:
"""Test various errors when registering a new device."""
await async_setup_component(hass, "http", {})
@@ -553,11 +554,11 @@ async def test_registering_new_device_validation(
resp = await client.post(REGISTER_URL, data=json.dumps({"browser": "chrome"}))
assert resp.status == HTTPStatus.BAD_REQUEST
- with patch("homeassistant.components.html5.notify.save_json", return_value=False):
- resp = await client.post(
- REGISTER_URL,
- data=json.dumps({"browser": "chrome", "subscription": "sub info"}),
- )
+ mock_save.return_value = False
+ resp = await client.post(
+ REGISTER_URL,
+ data=json.dumps({"browser": "chrome", "subscription": "sub info"}),
+ )
assert resp.status == HTTPStatus.BAD_REQUEST
@@ -566,6 +567,7 @@ async def test_unregistering_device_view(
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
load_config: MagicMock,
+ mock_save: MagicMock,
) -> None:
"""Test that the HTML unregister view works."""
load_config.return_value = {
@@ -582,11 +584,10 @@ async def test_unregistering_device_view(
client = await hass_client()
- with patch("homeassistant.components.html5.notify.save_json") as mock_save:
- resp = await client.delete(
- REGISTER_URL,
- data=json.dumps({"subscription": SUBSCRIPTION_1["subscription"]}),
- )
+ resp = await client.delete(
+ REGISTER_URL,
+ data=json.dumps({"subscription": SUBSCRIPTION_1["subscription"]}),
+ )
assert resp.status == HTTPStatus.OK
assert len(mock_save.mock_calls) == 1
@@ -600,6 +601,7 @@ async def test_unregister_device_view_handle_unknown_subscription(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
+ mock_save: MagicMock,
) -> None:
"""Test that the HTML unregister view handles unknown subscriptions."""
await async_setup_component(hass, "http", {})
@@ -612,11 +614,10 @@ async def test_unregister_device_view_handle_unknown_subscription(
client = await hass_client()
- with patch("homeassistant.components.html5.notify.save_json") as mock_save:
- resp = await client.delete(
- REGISTER_URL,
- data=json.dumps({"subscription": SUBSCRIPTION_3["subscription"]}),
- )
+ resp = await client.delete(
+ REGISTER_URL,
+ data=json.dumps({"subscription": SUBSCRIPTION_3["subscription"]}),
+ )
assert resp.status == HTTPStatus.OK, resp.response
assert len(mock_save.mock_calls) == 0
@@ -627,6 +628,7 @@ async def test_unregistering_device_view_handles_save_error(
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
load_config: MagicMock,
+ mock_save: MagicMock,
) -> None:
"""Test that the HTML unregister view handles save errors."""
load_config.return_value = {
@@ -643,14 +645,12 @@ async def test_unregistering_device_view_handles_save_error(
client = await hass_client()
- with patch(
- "homeassistant.components.html5.notify.save_json",
- side_effect=HomeAssistantError(),
- ):
- resp = await client.delete(
- REGISTER_URL,
- data=json.dumps({"subscription": SUBSCRIPTION_1["subscription"]}),
- )
+ mock_save.side_effect = HomeAssistantError()
+
+ resp = await client.delete(
+ REGISTER_URL,
+ data=json.dumps({"subscription": SUBSCRIPTION_1["subscription"]}),
+ )
assert resp.status == HTTPStatus.INTERNAL_SERVER_ERROR, resp.response
@@ -785,6 +785,7 @@ async def test_send_fcm_expired(
config_entry: MockConfigEntry,
load_config: MagicMock,
mock_wp: AsyncMock,
+ mock_save: MagicMock,
) -> None:
"""Test that the FCM target is removed when expired."""
load_config.return_value = {"device": SUBSCRIPTION_5}
@@ -796,15 +797,13 @@ async def test_send_fcm_expired(
assert config_entry.state is ConfigEntryState.LOADED
mock_wp.send_async.return_value.status = 410
- with (
- patch("homeassistant.components.html5.notify.save_json") as mock_save,
- ):
- await hass.services.async_call(
- "notify",
- "html5",
- {"message": "Hello", "target": ["device"], "data": {"icon": "beer.png"}},
- blocking=True,
- )
+
+ await hass.services.async_call(
+ "notify",
+ "html5",
+ {"message": "Hello", "target": ["device"], "data": {"icon": "beer.png"}},
+ blocking=True,
+ )
# "device" should be removed when expired.
mock_save.assert_called_once_with(hass.config.path(html5.REGISTRATIONS_FILE), {})
@@ -817,6 +816,7 @@ async def test_send_fcm_expired_save_fails(
load_config: MagicMock,
caplog: pytest.LogCaptureFixture,
mock_wp: AsyncMock,
+ mock_save: MagicMock,
) -> None:
"""Test that the FCM target remains after expiry if save_json fails."""
load_config.return_value = {"device": SUBSCRIPTION_5}
@@ -828,18 +828,13 @@ async def test_send_fcm_expired_save_fails(
assert config_entry.state is ConfigEntryState.LOADED
mock_wp.send_async.return_value.status = 410
- with (
- patch(
- "homeassistant.components.html5.notify.save_json",
- side_effect=HomeAssistantError(),
- ),
- ):
- await hass.services.async_call(
- "notify",
- "html5",
- {"message": "Hello", "target": ["device"], "data": {"icon": "beer.png"}},
- blocking=True,
- )
+ mock_save.side_effect = HomeAssistantError
+ await hass.services.async_call(
+ "notify",
+ "html5",
+ {"message": "Hello", "target": ["device"], "data": {"icon": "beer.png"}},
+ blocking=True,
+ )
# "device" should still exist if save fails.
assert "Error saving registration" in caplog.text
@@ -974,6 +969,7 @@ async def test_send_message_save_fails(
webpush_async: AsyncMock,
load_config: MagicMock,
caplog: pytest.LogCaptureFixture,
+ mock_save: MagicMock,
) -> None:
"""Test sending a message with channel expired but saving registration fails."""
load_config.return_value = {"my-desktop": SUBSCRIPTION_1}
@@ -987,13 +983,8 @@ async def test_send_message_save_fails(
webpush_async.side_effect = (
WebPushException("", response=Mock(status=HTTPStatus.GONE)),
)
- with (
- patch(
- "homeassistant.components.html5.notify.save_json",
- side_effect=HomeAssistantError,
- ),
- pytest.raises(HomeAssistantError) as e,
- ):
+ mock_save.side_effect = HomeAssistantError
+ with pytest.raises(HomeAssistantError) as e:
await hass.services.async_call(
NOTIFY_DOMAIN,
SERVICE_SEND_MESSAGE,