mirror of
https://github.com/home-assistant/core.git
synced 2026-05-16 13:31:01 +01:00
Manu
1233 lines
40 KiB
Python
1233 lines
40 KiB
Python
"""Test HTML5 notify platform."""
|
|
|
|
from collections.abc import Generator
|
|
from http import HTTPStatus
|
|
import json
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, MagicMock, Mock, mock_open, patch
|
|
|
|
from aiohttp import ClientError
|
|
from aiohttp.hdrs import AUTHORIZATION
|
|
import pytest
|
|
from pywebpush import WebPushException
|
|
from syrupy.assertion import SnapshotAssertion
|
|
|
|
from homeassistant.components.html5 import DOMAIN, notify as html5
|
|
from homeassistant.components.html5.const import (
|
|
ATTR_ACTIONS,
|
|
ATTR_BADGE,
|
|
ATTR_DIR,
|
|
ATTR_ICON,
|
|
ATTR_IMAGE,
|
|
ATTR_LANG,
|
|
ATTR_RENOTIFY,
|
|
ATTR_REQUIRE_INTERACTION,
|
|
ATTR_SILENT,
|
|
ATTR_TAG,
|
|
ATTR_TIMESTAMP,
|
|
ATTR_TTL,
|
|
ATTR_URGENCY,
|
|
ATTR_VIBRATE,
|
|
SERVICE_DISMISS,
|
|
)
|
|
from homeassistant.components.html5.notify import ATTR_ACTION, ATTR_DISMISS, DEFAULT_TTL
|
|
from homeassistant.components.html5.services import SERVICE_DISMISS_MESSAGE
|
|
from homeassistant.components.notify import (
|
|
ATTR_DATA,
|
|
ATTR_MESSAGE,
|
|
ATTR_TARGET,
|
|
ATTR_TITLE,
|
|
DOMAIN as NOTIFY_DOMAIN,
|
|
SERVICE_SEND_MESSAGE,
|
|
)
|
|
from homeassistant.config_entries import ConfigEntryState
|
|
from homeassistant.const import (
|
|
ATTR_ENTITY_ID,
|
|
STATE_UNAVAILABLE,
|
|
STATE_UNKNOWN,
|
|
Platform,
|
|
)
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
from homeassistant.helpers import entity_registry as er, issue_registry as ir
|
|
from homeassistant.setup import async_setup_component
|
|
|
|
from tests.common import MockConfigEntry, snapshot_platform
|
|
from tests.typing import ClientSessionGenerator
|
|
|
|
CONFIG_FILE = "file.conf"
|
|
|
|
VAPID_CONF = {
|
|
"platform": "html5",
|
|
"vapid_pub_key": (
|
|
"BJMA2gDZEkHaXRhf1fhY_"
|
|
"QbKbhVIHlSJXI0bFyo0eJXnUPOjdgycCAbj-2bMKMKNKs"
|
|
"_rM8JoSnyKGCXAY2dbONI"
|
|
),
|
|
"vapid_prv_key": "ZwPgwKpESGuGLMZYU39vKgrekrWzCijo-LsBM3CZ9-c",
|
|
"vapid_email": "someone@example.com",
|
|
}
|
|
|
|
SUBSCRIPTION_1 = {
|
|
"browser": "chrome",
|
|
"subscription": {
|
|
"endpoint": "https://googleapis.com",
|
|
"keys": {"auth": "auth", "p256dh": "p256dh"},
|
|
},
|
|
}
|
|
SUBSCRIPTION_2 = {
|
|
"browser": "firefox",
|
|
"subscription": {
|
|
"endpoint": "https://example.com",
|
|
"keys": {"auth": "bla", "p256dh": "bla"},
|
|
},
|
|
}
|
|
SUBSCRIPTION_3 = {
|
|
"browser": "chrome",
|
|
"subscription": {
|
|
"endpoint": "https://example.com/not_exist",
|
|
"keys": {"auth": "bla", "p256dh": "bla"},
|
|
},
|
|
}
|
|
SUBSCRIPTION_4 = {
|
|
"browser": "chrome",
|
|
"subscription": {
|
|
"endpoint": "https://googleapis.com",
|
|
"expirationTime": None,
|
|
"keys": {"auth": "auth", "p256dh": "p256dh"},
|
|
},
|
|
}
|
|
|
|
SUBSCRIPTION_5 = {
|
|
"browser": "chrome",
|
|
"subscription": {
|
|
"endpoint": "https://fcm.googleapis.com/fcm/send/LONG-RANDOM-KEY",
|
|
"expirationTime": None,
|
|
"keys": {"auth": "auth", "p256dh": "p256dh"},
|
|
},
|
|
}
|
|
|
|
REGISTER_URL = "/api/notify.html5"
|
|
PUBLISH_URL = "/api/notify.html5/callback"
|
|
|
|
VAPID_HEADERS = {
|
|
"Authorization": "vapid t=signed!!!",
|
|
"urgency": "normal",
|
|
"priority": "normal",
|
|
}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def notify_only() -> Generator[None]:
|
|
"""Enable only the notify platform."""
|
|
with patch(
|
|
"homeassistant.components.html5.PLATFORMS",
|
|
[Platform.NOTIFY],
|
|
):
|
|
yield
|
|
|
|
|
|
async def test_get_service_with_no_json(hass: HomeAssistant) -> None:
|
|
"""Test empty json file."""
|
|
await async_setup_component(hass, "http", {})
|
|
m = mock_open()
|
|
with patch("homeassistant.util.json.open", m, create=True):
|
|
service = await html5.async_get_service(hass, {}, VAPID_CONF)
|
|
|
|
assert service is not None
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_jwt", "mock_vapid")
|
|
@pytest.mark.freeze_time("2009-02-13T23:31:30.000Z")
|
|
async def test_dismissing_message(mock_wp: AsyncMock, hass: HomeAssistant) -> None:
|
|
"""Test dismissing message."""
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
data = {"device": SUBSCRIPTION_1}
|
|
|
|
m = mock_open(read_data=json.dumps(data))
|
|
with patch("homeassistant.util.json.open", m, create=True):
|
|
service = await html5.async_get_service(hass, {}, VAPID_CONF)
|
|
service.hass = hass
|
|
|
|
assert service is not None
|
|
|
|
await service.async_dismiss(target=["device", "non_existing"], data={"tag": "test"})
|
|
|
|
mock_wp.send_async.assert_awaited_once_with(
|
|
data='{"tag": "test", "dismiss": true, "data": {"jwt": "JWT"}, "timestamp": 1234567890000}',
|
|
headers=VAPID_HEADERS,
|
|
ttl=86400,
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_jwt", "mock_vapid", "mock_uuid")
|
|
@pytest.mark.freeze_time("2009-02-13T23:31:30.000Z")
|
|
async def test_sending_message(mock_wp: AsyncMock, hass: HomeAssistant) -> None:
|
|
"""Test sending message."""
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
data = {"device": SUBSCRIPTION_1}
|
|
|
|
m = mock_open(read_data=json.dumps(data))
|
|
with patch("homeassistant.util.json.open", m, create=True):
|
|
service = await html5.async_get_service(hass, {}, VAPID_CONF)
|
|
service.hass = hass
|
|
|
|
assert service is not None
|
|
|
|
await service.async_send_message(
|
|
"Hello", target=["device", "non_existing"], data={"icon": "beer.png"}
|
|
)
|
|
|
|
mock_wp.send_async.assert_awaited_once_with(
|
|
data='{"badge": "/static/images/notification-badge.png", "body": "Hello", "data": {"url": "/", "jwt": "JWT"}, "icon": "beer.png", "tag": "12345678-1234-5678-1234-567812345678", "title": "Home Assistant", "timestamp": 1234567890000}',
|
|
headers=VAPID_HEADERS,
|
|
ttl=86400,
|
|
)
|
|
|
|
# WebPusher constructor
|
|
assert mock_wp.cls.call_args[0][0] == SUBSCRIPTION_1["subscription"]
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_jwt", "mock_vapid", "mock_uuid")
|
|
@pytest.mark.freeze_time("2009-02-13T23:31:30.000Z")
|
|
async def test_fcm_key_include(mock_wp: AsyncMock, hass: HomeAssistant) -> None:
|
|
"""Test if the FCM header is included."""
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
data = {"chrome": SUBSCRIPTION_5}
|
|
|
|
m = mock_open(read_data=json.dumps(data))
|
|
with patch("homeassistant.util.json.open", m, create=True):
|
|
service = await html5.async_get_service(hass, {}, VAPID_CONF)
|
|
service.hass = hass
|
|
|
|
assert service is not None
|
|
|
|
await service.async_send_message("Hello", target=["chrome"])
|
|
|
|
mock_wp.send_async.assert_awaited_once_with(
|
|
data='{"badge": "/static/images/notification-badge.png", "body": "Hello", "data": {"url": "/", "jwt": "JWT"}, "icon": "/static/icons/favicon-192x192.png", "tag": "12345678-1234-5678-1234-567812345678", "title": "Home Assistant", "timestamp": 1234567890000}',
|
|
headers=VAPID_HEADERS,
|
|
ttl=86400,
|
|
)
|
|
|
|
# WebPusher constructor
|
|
assert mock_wp.cls.call_args[0][0] == SUBSCRIPTION_5["subscription"]
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_jwt", "mock_vapid", "mock_uuid")
|
|
@pytest.mark.freeze_time("2009-02-13T23:31:30.000Z")
|
|
async def test_fcm_send_with_unknown_priority(
|
|
mock_wp: AsyncMock, hass: HomeAssistant
|
|
) -> None:
|
|
"""Test if the gcm_key is only included for GCM endpoints."""
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
data = {"chrome": SUBSCRIPTION_5}
|
|
|
|
m = mock_open(read_data=json.dumps(data))
|
|
with patch("homeassistant.util.json.open", m, create=True):
|
|
service = await html5.async_get_service(hass, {}, VAPID_CONF)
|
|
service.hass = hass
|
|
|
|
assert service is not None
|
|
|
|
await service.async_send_message("Hello", target=["chrome"], priority="undefined")
|
|
|
|
mock_wp.send_async.assert_awaited_once_with(
|
|
data='{"badge": "/static/images/notification-badge.png", "body": "Hello", "data": {"url": "/", "jwt": "JWT"}, "icon": "/static/icons/favicon-192x192.png", "tag": "12345678-1234-5678-1234-567812345678", "title": "Home Assistant", "timestamp": 1234567890000}',
|
|
headers=VAPID_HEADERS,
|
|
ttl=86400,
|
|
)
|
|
# WebPusher constructor
|
|
assert mock_wp.cls.call_args[0][0] == SUBSCRIPTION_5["subscription"]
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_jwt", "mock_vapid", "mock_uuid")
|
|
@pytest.mark.freeze_time("2009-02-13T23:31:30.000Z")
|
|
async def test_fcm_no_targets(mock_wp: AsyncMock, hass: HomeAssistant) -> None:
|
|
"""Test if the gcm_key is only included for GCM endpoints."""
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
data = {"chrome": SUBSCRIPTION_5}
|
|
|
|
m = mock_open(read_data=json.dumps(data))
|
|
with patch("homeassistant.util.json.open", m, create=True):
|
|
service = await html5.async_get_service(hass, {}, VAPID_CONF)
|
|
service.hass = hass
|
|
|
|
assert service is not None
|
|
|
|
await service.async_send_message("Hello")
|
|
|
|
mock_wp.send_async.assert_awaited_once_with(
|
|
data='{"badge": "/static/images/notification-badge.png", "body": "Hello", "data": {"url": "/", "jwt": "JWT"}, "icon": "/static/icons/favicon-192x192.png", "tag": "12345678-1234-5678-1234-567812345678", "title": "Home Assistant", "timestamp": 1234567890000}',
|
|
headers=VAPID_HEADERS,
|
|
ttl=86400,
|
|
)
|
|
# WebPusher constructor
|
|
assert mock_wp.cls.call_args[0][0] == SUBSCRIPTION_5["subscription"]
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_jwt", "mock_vapid", "mock_uuid")
|
|
@pytest.mark.freeze_time("2009-02-13T23:31:30.000Z")
|
|
async def test_fcm_additional_data(mock_wp: AsyncMock, hass: HomeAssistant) -> None:
|
|
"""Test if the gcm_key is only included for GCM endpoints."""
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
data = {"chrome": SUBSCRIPTION_5}
|
|
|
|
m = mock_open(read_data=json.dumps(data))
|
|
with patch("homeassistant.util.json.open", m, create=True):
|
|
service = await html5.async_get_service(hass, {}, VAPID_CONF)
|
|
service.hass = hass
|
|
|
|
assert service is not None
|
|
|
|
await service.async_send_message("Hello", data={"mykey": "myvalue"})
|
|
|
|
mock_wp.send_async.assert_awaited_once_with(
|
|
data='{"badge": "/static/images/notification-badge.png", "body": "Hello", "data": {"mykey": "myvalue", "url": "/", "jwt": "JWT"}, "icon": "/static/icons/favicon-192x192.png", "tag": "12345678-1234-5678-1234-567812345678", "title": "Home Assistant", "timestamp": 1234567890000}',
|
|
headers=VAPID_HEADERS,
|
|
ttl=86400,
|
|
)
|
|
# WebPusher constructor
|
|
assert mock_wp.cls.call_args[0][0] == SUBSCRIPTION_5["subscription"]
|
|
|
|
|
|
@pytest.mark.usefixtures("load_config")
|
|
async def test_registering_new_device_view(
|
|
hass: HomeAssistant,
|
|
hass_client: ClientSessionGenerator,
|
|
config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""Test that the HTML view works."""
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
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))
|
|
|
|
assert resp.status == HTTPStatus.OK
|
|
assert len(mock_save.mock_calls) == 1
|
|
assert mock_save.mock_calls[0][1][1] == {"unnamed device": SUBSCRIPTION_1}
|
|
|
|
|
|
@pytest.mark.usefixtures("load_config")
|
|
async def test_registering_new_device_view_with_name(
|
|
hass: HomeAssistant,
|
|
hass_client: ClientSessionGenerator,
|
|
config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""Test that the HTML view works with name attribute."""
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
client = await hass_client()
|
|
|
|
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))
|
|
|
|
assert resp.status == HTTPStatus.OK
|
|
assert len(mock_save.mock_calls) == 1
|
|
assert mock_save.mock_calls[0][1][1] == {"test device": SUBSCRIPTION_1}
|
|
|
|
|
|
@pytest.mark.usefixtures("load_config")
|
|
async def test_registering_new_device_expiration_view(
|
|
hass: HomeAssistant,
|
|
hass_client: ClientSessionGenerator,
|
|
config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""Test that the HTML view works."""
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
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))
|
|
|
|
assert resp.status == HTTPStatus.OK
|
|
assert mock_save.mock_calls[0][1][1] == {"unnamed device": SUBSCRIPTION_4}
|
|
|
|
|
|
@pytest.mark.usefixtures("load_config")
|
|
async def test_registering_new_device_fails_view(
|
|
hass: HomeAssistant,
|
|
hass_client: ClientSessionGenerator,
|
|
config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""Test subs. are not altered when registering a new device fails."""
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
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))
|
|
|
|
assert resp.status == HTTPStatus.INTERNAL_SERVER_ERROR
|
|
|
|
|
|
@pytest.mark.usefixtures("load_config")
|
|
async def test_registering_existing_device_view(
|
|
hass: HomeAssistant,
|
|
hass_client: ClientSessionGenerator,
|
|
config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""Test subscription is updated when registering existing device."""
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
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))
|
|
|
|
assert resp.status == HTTPStatus.OK
|
|
mock_save.assert_called_with(
|
|
hass.config.path(html5.REGISTRATIONS_FILE), {"unnamed device": SUBSCRIPTION_4}
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("load_config")
|
|
async def test_registering_existing_device_view_with_name(
|
|
hass: HomeAssistant,
|
|
hass_client: ClientSessionGenerator,
|
|
config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""Test subscription is updated when reg'ing existing device with name."""
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
client = await hass_client()
|
|
|
|
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))
|
|
|
|
assert resp.status == HTTPStatus.OK
|
|
|
|
mock_save.assert_called_with(
|
|
hass.config.path(html5.REGISTRATIONS_FILE), {"test device": SUBSCRIPTION_4}
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("load_config")
|
|
async def test_registering_existing_device_fails_view(
|
|
hass: HomeAssistant,
|
|
hass_client: ClientSessionGenerator,
|
|
config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""Test sub. is not updated when registering existing device fails."""
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
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))
|
|
|
|
assert resp.status == HTTPStatus.INTERNAL_SERVER_ERROR
|
|
|
|
|
|
@pytest.mark.usefixtures("load_config")
|
|
async def test_registering_new_device_validation(
|
|
hass: HomeAssistant,
|
|
hass_client: ClientSessionGenerator,
|
|
config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""Test various errors when registering a new device."""
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
client = await hass_client()
|
|
|
|
resp = await client.post(
|
|
REGISTER_URL,
|
|
data=json.dumps({"browser": "invalid browser", "subscription": "sub info"}),
|
|
)
|
|
assert resp.status == HTTPStatus.BAD_REQUEST
|
|
|
|
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"}),
|
|
)
|
|
assert resp.status == HTTPStatus.BAD_REQUEST
|
|
|
|
|
|
async def test_unregistering_device_view(
|
|
hass: HomeAssistant,
|
|
hass_client: ClientSessionGenerator,
|
|
config_entry: MockConfigEntry,
|
|
load_config: MagicMock,
|
|
) -> None:
|
|
"""Test that the HTML unregister view works."""
|
|
load_config.return_value = {
|
|
"some device": SUBSCRIPTION_1,
|
|
"other device": SUBSCRIPTION_2,
|
|
}
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
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"]}),
|
|
)
|
|
|
|
assert resp.status == HTTPStatus.OK
|
|
assert len(mock_save.mock_calls) == 1
|
|
mock_save.assert_called_once_with(
|
|
hass.config.path(html5.REGISTRATIONS_FILE), {"other device": SUBSCRIPTION_2}
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("load_config")
|
|
async def test_unregister_device_view_handle_unknown_subscription(
|
|
hass: HomeAssistant,
|
|
hass_client: ClientSessionGenerator,
|
|
config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""Test that the HTML unregister view handles unknown subscriptions."""
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
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"]}),
|
|
)
|
|
|
|
assert resp.status == HTTPStatus.OK, resp.response
|
|
assert len(mock_save.mock_calls) == 0
|
|
|
|
|
|
async def test_unregistering_device_view_handles_save_error(
|
|
hass: HomeAssistant,
|
|
hass_client: ClientSessionGenerator,
|
|
config_entry: MockConfigEntry,
|
|
load_config: MagicMock,
|
|
) -> None:
|
|
"""Test that the HTML unregister view handles save errors."""
|
|
load_config.return_value = {
|
|
"some device": SUBSCRIPTION_1,
|
|
"other device": SUBSCRIPTION_2,
|
|
}
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
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.delete(
|
|
REGISTER_URL,
|
|
data=json.dumps({"subscription": SUBSCRIPTION_1["subscription"]}),
|
|
)
|
|
|
|
assert resp.status == HTTPStatus.INTERNAL_SERVER_ERROR, resp.response
|
|
|
|
|
|
@pytest.mark.usefixtures("load_config")
|
|
async def test_callback_view_no_jwt(
|
|
hass: HomeAssistant,
|
|
hass_client: ClientSessionGenerator,
|
|
config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""Test that the notification callback view works without JWT."""
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
client = await hass_client()
|
|
resp = await client.post(
|
|
PUBLISH_URL,
|
|
data=json.dumps(
|
|
{"type": "push", "tag": "3bc28d69-0921-41f1-ac6a-7a627ba0aa72"}
|
|
),
|
|
)
|
|
|
|
assert resp.status == HTTPStatus.UNAUTHORIZED
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_jwt", "mock_vapid", "mock_uuid")
|
|
@pytest.mark.freeze_time("2009-02-13T23:31:30.000Z")
|
|
async def test_callback_view_with_jwt(
|
|
hass: HomeAssistant,
|
|
hass_client: ClientSessionGenerator,
|
|
config_entry: MockConfigEntry,
|
|
load_config: MagicMock,
|
|
mock_wp: AsyncMock,
|
|
) -> None:
|
|
"""Test that the notification callback view works with JWT."""
|
|
load_config.return_value = {"device": SUBSCRIPTION_1}
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
client = await hass_client()
|
|
|
|
await hass.services.async_call(
|
|
"notify",
|
|
"html5",
|
|
{"message": "Hello", "target": ["device"], "data": {"icon": "beer.png"}},
|
|
blocking=True,
|
|
)
|
|
|
|
mock_wp.send_async.assert_awaited_once_with(
|
|
data='{"badge": "/static/images/notification-badge.png", "body": "Hello", "data": {"url": "/", "jwt": "JWT"}, "icon": "beer.png", "tag": "12345678-1234-5678-1234-567812345678", "title": "Home Assistant", "timestamp": 1234567890000}',
|
|
headers=VAPID_HEADERS,
|
|
ttl=86400,
|
|
)
|
|
# WebPusher constructor
|
|
assert mock_wp.cls.call_args[0][0] == SUBSCRIPTION_1["subscription"]
|
|
|
|
bearer_token = "Bearer JWT"
|
|
|
|
resp = await client.post(
|
|
PUBLISH_URL, json={"type": "push"}, headers={AUTHORIZATION: bearer_token}
|
|
)
|
|
|
|
assert resp.status == HTTPStatus.OK
|
|
body = await resp.json()
|
|
assert body == {"event": "push", "status": "ok"}
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_jwt", "mock_vapid", "mock_uuid")
|
|
@pytest.mark.freeze_time("2009-02-13T23:31:30.000Z")
|
|
async def test_send_fcm_without_targets(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
load_config: MagicMock,
|
|
mock_wp: AsyncMock,
|
|
) -> None:
|
|
"""Test that the notification is send with FCM without targets."""
|
|
load_config.return_value = {"device": SUBSCRIPTION_5}
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
await hass.services.async_call(
|
|
"notify",
|
|
"html5",
|
|
{"message": "Hello", "target": ["device"], "data": {"icon": "beer.png"}},
|
|
blocking=True,
|
|
)
|
|
|
|
mock_wp.send_async.assert_awaited_once_with(
|
|
data='{"badge": "/static/images/notification-badge.png", "body": "Hello", "data": {"url": "/", "jwt": "JWT"}, "icon": "beer.png", "tag": "12345678-1234-5678-1234-567812345678", "title": "Home Assistant", "timestamp": 1234567890000}',
|
|
headers=VAPID_HEADERS,
|
|
ttl=86400,
|
|
)
|
|
# WebPusher constructor
|
|
assert mock_wp.cls.call_args[0][0] == SUBSCRIPTION_5["subscription"]
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_jwt", "mock_vapid", "mock_uuid")
|
|
@pytest.mark.freeze_time("2009-02-13T23:31:30.000Z")
|
|
async def test_send_fcm_expired(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
load_config: MagicMock,
|
|
mock_wp: AsyncMock,
|
|
) -> None:
|
|
"""Test that the FCM target is removed when expired."""
|
|
load_config.return_value = {"device": SUBSCRIPTION_5}
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
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,
|
|
)
|
|
# "device" should be removed when expired.
|
|
mock_save.assert_called_once_with(hass.config.path(html5.REGISTRATIONS_FILE), {})
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_jwt", "mock_vapid", "mock_uuid")
|
|
@pytest.mark.freeze_time("2009-02-13T23:31:30.000Z")
|
|
async def test_send_fcm_expired_save_fails(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
load_config: MagicMock,
|
|
caplog: pytest.LogCaptureFixture,
|
|
mock_wp: AsyncMock,
|
|
) -> None:
|
|
"""Test that the FCM target remains after expiry if save_json fails."""
|
|
load_config.return_value = {"device": SUBSCRIPTION_5}
|
|
await async_setup_component(hass, "http", {})
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
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,
|
|
)
|
|
# "device" should still exist if save fails.
|
|
assert "Error saving registration" in caplog.text
|
|
|
|
|
|
async def test_notify_platform(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
snapshot: SnapshotAssertion,
|
|
entity_registry: er.EntityRegistry,
|
|
load_config: MagicMock,
|
|
) -> None:
|
|
"""Test setup of the notify platform."""
|
|
load_config.return_value = {"my-desktop": SUBSCRIPTION_1}
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
await snapshot_platform(hass, entity_registry, snapshot, config_entry.entry_id)
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_jwt", "mock_vapid", "mock_uuid")
|
|
@pytest.mark.freeze_time("2009-02-13T23:31:30.000Z")
|
|
async def test_send_message(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
webpush_async: AsyncMock,
|
|
load_config: MagicMock,
|
|
) -> None:
|
|
"""Test sending a message."""
|
|
load_config.return_value = {"my-desktop": SUBSCRIPTION_1}
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
state = hass.states.get("notify.my_desktop")
|
|
assert state
|
|
assert state.state == STATE_UNKNOWN
|
|
|
|
await hass.services.async_call(
|
|
NOTIFY_DOMAIN,
|
|
SERVICE_SEND_MESSAGE,
|
|
{
|
|
ATTR_ENTITY_ID: "notify.my_desktop",
|
|
ATTR_MESSAGE: "World",
|
|
ATTR_TITLE: "Hello",
|
|
},
|
|
blocking=True,
|
|
)
|
|
|
|
state = hass.states.get("notify.my_desktop")
|
|
assert state
|
|
assert state.state == "2009-02-13T23:31:30+00:00"
|
|
|
|
webpush_async.assert_awaited_once()
|
|
assert webpush_async.await_args
|
|
_, payload, _, _ = webpush_async.await_args.args
|
|
assert json.loads(payload) == {
|
|
"title": "Hello",
|
|
"body": "World",
|
|
"badge": "/static/images/notification-badge.png",
|
|
"icon": "/static/icons/favicon-192x192.png",
|
|
"tag": "12345678-1234-5678-1234-567812345678",
|
|
"timestamp": 1234567890000,
|
|
"data": {"jwt": "JWT"},
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("exception", "translation_key"),
|
|
[
|
|
(
|
|
WebPushException("", response=Mock(status=HTTPStatus.IM_A_TEAPOT)),
|
|
"request_error",
|
|
),
|
|
(
|
|
WebPushException("", response=Mock(status=HTTPStatus.GONE)),
|
|
"channel_expired",
|
|
),
|
|
(
|
|
ClientError,
|
|
"connection_error",
|
|
),
|
|
],
|
|
)
|
|
@pytest.mark.parametrize("domain", [NOTIFY_DOMAIN, DOMAIN])
|
|
@pytest.mark.usefixtures("mock_jwt", "mock_vapid", "mock_uuid")
|
|
@pytest.mark.freeze_time("2009-02-13T23:31:30.000Z")
|
|
async def test_send_message_exceptions(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
webpush_async: AsyncMock,
|
|
load_config: MagicMock,
|
|
exception: Exception,
|
|
translation_key: str,
|
|
domain: str,
|
|
) -> None:
|
|
"""Test sending a message with exceptions."""
|
|
load_config.return_value = {"my-desktop": SUBSCRIPTION_1}
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
webpush_async.side_effect = exception
|
|
|
|
with pytest.raises(HomeAssistantError) as e:
|
|
await hass.services.async_call(
|
|
domain,
|
|
SERVICE_SEND_MESSAGE,
|
|
{
|
|
ATTR_ENTITY_ID: "notify.my_desktop",
|
|
ATTR_MESSAGE: "World",
|
|
ATTR_TITLE: "Hello",
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert e.value.translation_key == translation_key
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_jwt", "mock_vapid", "mock_uuid")
|
|
@pytest.mark.freeze_time("2009-02-13T23:31:30.000Z")
|
|
async def test_send_message_save_fails(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
webpush_async: AsyncMock,
|
|
load_config: MagicMock,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""Test sending a message with channel expired but saving registration fails."""
|
|
load_config.return_value = {"my-desktop": SUBSCRIPTION_1}
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
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,
|
|
):
|
|
await hass.services.async_call(
|
|
NOTIFY_DOMAIN,
|
|
SERVICE_SEND_MESSAGE,
|
|
{
|
|
ATTR_ENTITY_ID: "notify.my_desktop",
|
|
ATTR_MESSAGE: "World",
|
|
ATTR_TITLE: "Hello",
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert e.value.translation_key == "channel_expired"
|
|
|
|
assert "Error saving registration" in caplog.text
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_jwt", "mock_vapid", "mock_uuid")
|
|
@pytest.mark.freeze_time("2009-02-13T23:31:30.000Z")
|
|
async def test_send_message_unavailable(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
webpush_async: AsyncMock,
|
|
load_config: MagicMock,
|
|
) -> None:
|
|
"""Test sending a message with channel expired and entity goes unavailable."""
|
|
load_config.return_value = {"my-desktop": SUBSCRIPTION_1}
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
webpush_async.side_effect = (
|
|
WebPushException("", response=Mock(status=HTTPStatus.GONE)),
|
|
)
|
|
with pytest.raises(HomeAssistantError) as e:
|
|
await hass.services.async_call(
|
|
NOTIFY_DOMAIN,
|
|
SERVICE_SEND_MESSAGE,
|
|
{
|
|
ATTR_ENTITY_ID: "notify.my_desktop",
|
|
ATTR_MESSAGE: "World",
|
|
ATTR_TITLE: "Hello",
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert e.value.translation_key == "channel_expired"
|
|
|
|
state = hass.states.get("notify.my_desktop")
|
|
assert state
|
|
assert state.state == STATE_UNAVAILABLE
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("service_data", "expected_payload", "expected_ttl", "expected_headers"),
|
|
[
|
|
({ATTR_MESSAGE: "World"}, {"body": "World"}, DEFAULT_TTL, None),
|
|
(
|
|
{ATTR_ICON: "/static/icons/favicon-192x192.png"},
|
|
{"icon": "/static/icons/favicon-192x192.png"},
|
|
DEFAULT_TTL,
|
|
None,
|
|
),
|
|
(
|
|
{ATTR_BADGE: "/static/images/notification-badge.png"},
|
|
{"badge": "/static/images/notification-badge.png"},
|
|
DEFAULT_TTL,
|
|
None,
|
|
),
|
|
(
|
|
{ATTR_IMAGE: "/static/images/image.jpg"},
|
|
{"image": "/static/images/image.jpg"},
|
|
DEFAULT_TTL,
|
|
None,
|
|
),
|
|
({ATTR_TAG: "message-group-1"}, {"tag": "message-group-1"}, DEFAULT_TTL, None),
|
|
({ATTR_DIR: "rtl"}, {"dir": "rtl"}, DEFAULT_TTL, None),
|
|
({ATTR_RENOTIFY: True}, {"renotify": True}, DEFAULT_TTL, None),
|
|
({ATTR_SILENT: True}, {"silent": True}, DEFAULT_TTL, None),
|
|
(
|
|
{ATTR_REQUIRE_INTERACTION: True},
|
|
{"requireInteraction": True},
|
|
DEFAULT_TTL,
|
|
None,
|
|
),
|
|
(
|
|
{ATTR_VIBRATE: [200, 100, 200]},
|
|
{"vibrate": [200, 100, 200]},
|
|
DEFAULT_TTL,
|
|
None,
|
|
),
|
|
({ATTR_LANG: "es-419"}, {"lang": "es-419"}, DEFAULT_TTL, None),
|
|
({ATTR_TIMESTAMP: "1970-01-01 00:00:00"}, {"timestamp": 0}, DEFAULT_TTL, None),
|
|
({ATTR_TTL: {"days": 28}}, {}, 2419200, None),
|
|
({ATTR_TTL: {"seconds": 0}}, {}, 0, None),
|
|
(
|
|
{ATTR_URGENCY: "high"},
|
|
{},
|
|
DEFAULT_TTL,
|
|
{"Urgency": "high"},
|
|
),
|
|
(
|
|
{
|
|
ATTR_ACTIONS: [
|
|
{
|
|
ATTR_ACTION: "callback-event",
|
|
ATTR_TITLE: "Callback Event",
|
|
ATTR_ICON: "/static/icons/favicon-192x192.png",
|
|
}
|
|
]
|
|
},
|
|
{
|
|
"actions": [
|
|
{
|
|
"action": "callback-event",
|
|
"title": "Callback Event",
|
|
"icon": "/static/icons/favicon-192x192.png",
|
|
}
|
|
]
|
|
},
|
|
DEFAULT_TTL,
|
|
None,
|
|
),
|
|
(
|
|
{ATTR_DATA: {"customKey": "customValue"}},
|
|
{"data": {"jwt": "JWT", "customKey": "customValue"}},
|
|
DEFAULT_TTL,
|
|
None,
|
|
),
|
|
],
|
|
)
|
|
@pytest.mark.usefixtures("mock_jwt", "mock_vapid", "mock_uuid")
|
|
@pytest.mark.freeze_time("2009-02-13T23:31:30.000Z")
|
|
async def test_html5_send_message(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
webpush_async: AsyncMock,
|
|
load_config: MagicMock,
|
|
service_data: dict[str, Any],
|
|
expected_payload: dict[str, Any],
|
|
expected_ttl: int,
|
|
expected_headers: dict[str, Any] | None,
|
|
) -> None:
|
|
"""Test sending a message via html5.send_message action."""
|
|
load_config.return_value = {"my-desktop": SUBSCRIPTION_1}
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
state = hass.states.get("notify.my_desktop")
|
|
assert state
|
|
assert state.state == STATE_UNKNOWN
|
|
|
|
await hass.services.async_call(
|
|
DOMAIN,
|
|
SERVICE_SEND_MESSAGE,
|
|
{ATTR_ENTITY_ID: "notify.my_desktop", ATTR_TITLE: "Hello", **service_data},
|
|
blocking=True,
|
|
)
|
|
|
|
state = hass.states.get("notify.my_desktop")
|
|
assert state
|
|
assert state.state == "2009-02-13T23:31:30+00:00"
|
|
|
|
webpush_async.assert_awaited_once()
|
|
assert webpush_async.await_args
|
|
_, payload, _, _ = webpush_async.await_args.args
|
|
assert json.loads(payload) == {
|
|
"title": "Hello",
|
|
"tag": "12345678-1234-5678-1234-567812345678",
|
|
"timestamp": 1234567890000,
|
|
"data": {"jwt": "JWT"},
|
|
**expected_payload,
|
|
}
|
|
|
|
assert webpush_async.await_args.kwargs["ttl"] == expected_ttl
|
|
assert webpush_async.await_args.kwargs["headers"] == expected_headers
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("domain", "service", "service_data", "issue_id"),
|
|
[
|
|
(
|
|
NOTIFY_DOMAIN,
|
|
"html5_my_desktop",
|
|
{ATTR_MESSAGE: "Hello", ATTR_TARGET: ["my-desktop"]},
|
|
"deprecated_notify_action_notify.html5_my_desktop",
|
|
),
|
|
(
|
|
NOTIFY_DOMAIN,
|
|
DOMAIN,
|
|
{ATTR_MESSAGE: "Hello"},
|
|
"deprecated_notify_action_notify.html5",
|
|
),
|
|
(
|
|
NOTIFY_DOMAIN,
|
|
DOMAIN,
|
|
{ATTR_MESSAGE: "Hello", ATTR_TARGET: ["my-desktop", "my-phone"]},
|
|
"deprecated_notify_action_notify.html5",
|
|
),
|
|
(
|
|
DOMAIN,
|
|
SERVICE_DISMISS,
|
|
{},
|
|
"deprecated_dismiss_action",
|
|
),
|
|
],
|
|
)
|
|
@pytest.mark.usefixtures("mock_wp", "mock_jwt", "mock_vapid", "mock_uuid")
|
|
async def test_deprecation_action_call(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
load_config: MagicMock,
|
|
issue_registry: ir.IssueRegistry,
|
|
domain: str,
|
|
service: str,
|
|
service_data: dict[str, Any] | None,
|
|
issue_id: str,
|
|
) -> None:
|
|
"""Test deprecation action call."""
|
|
load_config.return_value = {
|
|
"my-desktop": SUBSCRIPTION_1,
|
|
"my-phone": SUBSCRIPTION_2,
|
|
}
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
await hass.services.async_call(
|
|
domain,
|
|
service,
|
|
service_data,
|
|
blocking=True,
|
|
)
|
|
|
|
assert issue_registry.async_get_issue(
|
|
domain=DOMAIN,
|
|
issue_id=issue_id,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("service_data", "expected_payload"),
|
|
[
|
|
(
|
|
{ATTR_TAG: "message-group-1"},
|
|
{ATTR_DISMISS: True, ATTR_TAG: "message-group-1"},
|
|
),
|
|
(
|
|
{},
|
|
{ATTR_DISMISS: True, ATTR_TAG: ""},
|
|
),
|
|
],
|
|
)
|
|
@pytest.mark.usefixtures("mock_jwt", "mock_vapid", "mock_uuid")
|
|
@pytest.mark.freeze_time("2009-02-13T23:31:30.000Z")
|
|
async def test_html5_dismiss_message(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
webpush_async: AsyncMock,
|
|
load_config: MagicMock,
|
|
service_data: dict[str, Any],
|
|
expected_payload: dict[str, Any],
|
|
) -> None:
|
|
"""Test dismissing a message via html5.dismiss_message action."""
|
|
load_config.return_value = {"my-desktop": SUBSCRIPTION_1}
|
|
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert config_entry.state is ConfigEntryState.LOADED
|
|
|
|
await hass.services.async_call(
|
|
DOMAIN,
|
|
SERVICE_DISMISS_MESSAGE,
|
|
{
|
|
ATTR_ENTITY_ID: "notify.my_desktop",
|
|
**service_data,
|
|
},
|
|
blocking=True,
|
|
)
|
|
|
|
webpush_async.assert_awaited_once()
|
|
assert webpush_async.await_args
|
|
_, payload, _, _ = webpush_async.await_args.args
|
|
assert json.loads(payload) == {
|
|
"timestamp": 1234567890000,
|
|
"data": {"jwt": "JWT"},
|
|
**expected_payload,
|
|
}
|