From e04ecc69120553c6caaf7ba78fe7baed6404d21f Mon Sep 17 00:00:00 2001 From: Manu Date: Sun, 13 Sep 2026 11:40:50 +0200 Subject: [PATCH] Move HTTP views into their own module in HTML5 integration (#181889) --- homeassistant/components/html5/const.py | 2 + homeassistant/components/html5/http.py | 302 +++++++++++++++++++++++ homeassistant/components/html5/notify.py | 280 +-------------------- tests/components/html5/conftest.py | 12 + tests/components/html5/test_event.py | 4 +- tests/components/html5/test_notify.py | 137 +++++----- 6 files changed, 387 insertions(+), 350 deletions(-) create mode 100644 homeassistant/components/html5/http.py 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,