Move HTTP views into their own module in HTML5 integration (#181889)

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"

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]})

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."""
 

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

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

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,