mirror of
https://github.com/home-assistant/core.git
synced 2025-04-24 09:17:53 +00:00
Signal messenger attachments as bytes support (#62311)
Co-authored-by: Alex <33379584+alexyao2015@users.noreply.github.com> Co-authored-by: Ian Byrne <ian.byrne@burnsie.com.au>
This commit is contained in:
Alan Byrne
GitHub
parent
250379e181
commit
a474c1e342
@ -2,7 +2,11 @@
|
||||
"domain": "signal_messenger",
|
||||
"name": "Signal Messenger",
|
||||
"documentation": "https://www.home-assistant.io/integrations/signal_messenger",
|
||||
"codeowners": ["@bbernhard"],
|
||||
"requirements": ["pysignalclirestapi==0.3.4"],
|
||||
"codeowners": [
|
||||
"@bbernhard"
|
||||
],
|
||||
"requirements": [
|
||||
"pysignalclirestapi==0.3.18"
|
||||
],
|
||||
"iot_class": "cloud_push"
|
||||
}
|
||||
}
|
||||
@ -1,7 +1,11 @@
|
||||
"""Signal Messenger for notify component."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from pysignalclirestapi import SignalCliRestApi, SignalCliRestApiError
|
||||
import requests
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.notify import (
|
||||
@ -9,15 +13,34 @@ from homeassistant.components.notify import (
|
||||
PLATFORM_SCHEMA,
|
||||
BaseNotificationService,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
CONF_SENDER_NR = "number"
|
||||
CONF_RECP_NR = "recipients"
|
||||
CONF_SIGNAL_CLI_REST_API = "url"
|
||||
ATTR_FILENAME = "attachment"
|
||||
CONF_MAX_ALLOWED_DOWNLOAD_SIZE_BYTES = 52428800
|
||||
ATTR_FILENAMES = "attachments"
|
||||
ATTR_URLS = "urls"
|
||||
ATTR_VERIFY_SSL = "verify_ssl"
|
||||
|
||||
DATA_FILENAMES_SCHEMA = vol.Schema({vol.Required(ATTR_FILENAMES): [cv.string]})
|
||||
|
||||
DATA_URLS_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(ATTR_URLS): [cv.url],
|
||||
vol.Optional(ATTR_VERIFY_SSL, default=True): cv.boolean,
|
||||
}
|
||||
)
|
||||
|
||||
DATA_SCHEMA = vol.Any(
|
||||
None,
|
||||
DATA_FILENAMES_SCHEMA,
|
||||
DATA_URLS_SCHEMA,
|
||||
)
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
@ -28,7 +51,11 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
)
|
||||
|
||||
|
||||
def get_service(hass, config, discovery_info=None):
|
||||
def get_service(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> SignalNotificationService:
|
||||
"""Get the SignalMessenger notification service."""
|
||||
|
||||
sender_nr = config[CONF_SENDER_NR]
|
||||
@ -37,43 +64,118 @@ def get_service(hass, config, discovery_info=None):
|
||||
|
||||
signal_cli_rest_api = SignalCliRestApi(signal_cli_rest_api_url, sender_nr)
|
||||
|
||||
return SignalNotificationService(recp_nrs, signal_cli_rest_api)
|
||||
return SignalNotificationService(hass, recp_nrs, signal_cli_rest_api)
|
||||
|
||||
|
||||
class SignalNotificationService(BaseNotificationService):
|
||||
"""Implement the notification service for SignalMessenger."""
|
||||
|
||||
def __init__(self, recp_nrs, signal_cli_rest_api):
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
recp_nrs: list[str],
|
||||
signal_cli_rest_api: SignalCliRestApi,
|
||||
) -> None:
|
||||
"""Initialize the service."""
|
||||
|
||||
self._hass = hass
|
||||
self._recp_nrs = recp_nrs
|
||||
self._signal_cli_rest_api = signal_cli_rest_api
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
"""Send a message to a one or more recipients.
|
||||
|
||||
Additionally a file can be attached.
|
||||
"""
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to a one or more recipients. Additionally a file can be attached."""
|
||||
|
||||
_LOGGER.debug("Sending signal message")
|
||||
|
||||
data = kwargs.get(ATTR_DATA)
|
||||
|
||||
filenames = None
|
||||
if data is not None:
|
||||
if ATTR_FILENAMES in data:
|
||||
filenames = data[ATTR_FILENAMES]
|
||||
if ATTR_FILENAME in data:
|
||||
_LOGGER.warning(
|
||||
"The 'attachment' option is deprecated, please replace it with 'attachments'. This option will become invalid in version 0.108"
|
||||
)
|
||||
if filenames is None:
|
||||
filenames = [data[ATTR_FILENAME]]
|
||||
else:
|
||||
filenames.append(data[ATTR_FILENAME])
|
||||
try:
|
||||
data = DATA_SCHEMA(data)
|
||||
except vol.Invalid as ex:
|
||||
_LOGGER.error("Invalid message data: %s", ex)
|
||||
raise ex
|
||||
|
||||
filenames = self.get_filenames(data)
|
||||
attachments_as_bytes = self.get_attachments_as_bytes(
|
||||
data, CONF_MAX_ALLOWED_DOWNLOAD_SIZE_BYTES, self._hass
|
||||
)
|
||||
|
||||
try:
|
||||
self._signal_cli_rest_api.send_message(message, self._recp_nrs, filenames)
|
||||
self._signal_cli_rest_api.send_message(
|
||||
message, self._recp_nrs, filenames, attachments_as_bytes
|
||||
)
|
||||
except SignalCliRestApiError as ex:
|
||||
_LOGGER.error("%s", ex)
|
||||
raise ex
|
||||
|
||||
@staticmethod
|
||||
def get_filenames(data: Any) -> list[str] | None:
|
||||
"""Extract attachment filenames from data."""
|
||||
try:
|
||||
data = DATA_FILENAMES_SCHEMA(data)
|
||||
except vol.Invalid:
|
||||
return None
|
||||
|
||||
return data[ATTR_FILENAMES]
|
||||
|
||||
@staticmethod
|
||||
def get_attachments_as_bytes(
|
||||
data: Any,
|
||||
attachment_size_limit: int,
|
||||
hass: HomeAssistant,
|
||||
) -> list[bytearray] | None:
|
||||
"""Retrieve attachments from URLs defined in data."""
|
||||
try:
|
||||
data = DATA_URLS_SCHEMA(data)
|
||||
except vol.Invalid:
|
||||
return None
|
||||
|
||||
urls = data[ATTR_URLS]
|
||||
|
||||
attachments_as_bytes: list[bytearray] = []
|
||||
|
||||
for url in urls:
|
||||
try:
|
||||
if not hass.config.is_allowed_external_url(url):
|
||||
_LOGGER.error("URL '%s' not in allow list", url)
|
||||
continue
|
||||
|
||||
resp = requests.get(
|
||||
url, verify=data[ATTR_VERIFY_SSL], timeout=10, stream=True
|
||||
)
|
||||
resp.raise_for_status()
|
||||
|
||||
if (
|
||||
resp.headers.get("Content-Length") is not None
|
||||
and int(str(resp.headers.get("Content-Length")))
|
||||
> attachment_size_limit
|
||||
):
|
||||
raise ValueError(
|
||||
"Attachment too large (Content-Length reports {}). Max size: {} bytes".format(
|
||||
int(str(resp.headers.get("Content-Length"))),
|
||||
CONF_MAX_ALLOWED_DOWNLOAD_SIZE_BYTES,
|
||||
)
|
||||
)
|
||||
|
||||
size = 0
|
||||
chunks = bytearray()
|
||||
for chunk in resp.iter_content(1024):
|
||||
size += len(chunk)
|
||||
if size > attachment_size_limit:
|
||||
raise ValueError(
|
||||
"Attachment too large (Stream reports {}). Max size: {} bytes".format(
|
||||
size, CONF_MAX_ALLOWED_DOWNLOAD_SIZE_BYTES
|
||||
)
|
||||
)
|
||||
|
||||
chunks.extend(chunk)
|
||||
|
||||
attachments_as_bytes.append(chunks)
|
||||
except Exception as ex:
|
||||
_LOGGER.error("%s", ex)
|
||||
raise ex
|
||||
|
||||
if not attachments_as_bytes:
|
||||
return None
|
||||
|
||||
return attachments_as_bytes
|
||||
|
||||
@ -1831,7 +1831,7 @@ pysher==1.0.1
|
||||
pysiaalarm==3.0.2
|
||||
|
||||
# homeassistant.components.signal_messenger
|
||||
pysignalclirestapi==0.3.4
|
||||
pysignalclirestapi==0.3.18
|
||||
|
||||
# homeassistant.components.sky_hub
|
||||
pyskyqhub==0.1.4
|
||||
|
||||
@ -1155,7 +1155,7 @@ pyserial==3.5
|
||||
pysiaalarm==3.0.2
|
||||
|
||||
# homeassistant.components.signal_messenger
|
||||
pysignalclirestapi==0.3.4
|
||||
pysignalclirestapi==0.3.18
|
||||
|
||||
# homeassistant.components.sma
|
||||
pysma==0.6.10
|
||||
|
||||
@ -3,37 +3,69 @@ from http import HTTPStatus
|
||||
|
||||
from pysignalclirestapi import SignalCliRestApi
|
||||
import pytest
|
||||
from requests_mock.mocker import Mocker
|
||||
|
||||
from homeassistant.components.signal_messenger.notify import SignalNotificationService
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def signal_notification_service():
|
||||
"""Set up signal notification service."""
|
||||
recipients = ["+435565656565"]
|
||||
number = "+43443434343"
|
||||
client = SignalCliRestApi("http://127.0.0.1:8080", number)
|
||||
return SignalNotificationService(recipients, client)
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
SIGNAL_SEND_PATH_SUFIX = "/v2/send"
|
||||
MESSAGE = "Testing Signal Messenger platform :)"
|
||||
CONTENT = b"TestContent"
|
||||
NUMBER_FROM = "+43443434343"
|
||||
NUMBERS_TO = ["+435565656565"]
|
||||
URL_ATTACHMENT = "http://127.0.0.1:8080/image.jpg"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def signal_requests_mock(requests_mock):
|
||||
"""Prepare signal service mock."""
|
||||
requests_mock.register_uri(
|
||||
"POST",
|
||||
"http://127.0.0.1:8080" + SIGNAL_SEND_PATH_SUFIX,
|
||||
status_code=HTTPStatus.CREATED,
|
||||
)
|
||||
requests_mock.register_uri(
|
||||
"GET",
|
||||
"http://127.0.0.1:8080/v1/about",
|
||||
status_code=HTTPStatus.OK,
|
||||
json={"versions": ["v1", "v2"]},
|
||||
)
|
||||
return requests_mock
|
||||
def signal_notification_service(hass: HomeAssistant) -> SignalNotificationService:
|
||||
"""Set up signal notification service."""
|
||||
hass.config.allowlist_external_urls.add(URL_ATTACHMENT)
|
||||
recipients = ["+435565656565"]
|
||||
number = "+43443434343"
|
||||
client = SignalCliRestApi("http://127.0.0.1:8080", number)
|
||||
return SignalNotificationService(hass, recipients, client)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def signal_requests_mock_factory(requests_mock: Mocker) -> Mocker:
|
||||
"""Create signal service mock from factory."""
|
||||
|
||||
def _signal_requests_mock_factory(
|
||||
success_send_result: bool = True, content_length_header: str = None
|
||||
) -> Mocker:
|
||||
requests_mock.register_uri(
|
||||
"GET",
|
||||
"http://127.0.0.1:8080/v1/about",
|
||||
status_code=HTTPStatus.OK,
|
||||
json={"versions": ["v1", "v2"]},
|
||||
)
|
||||
if success_send_result:
|
||||
requests_mock.register_uri(
|
||||
"POST",
|
||||
"http://127.0.0.1:8080" + SIGNAL_SEND_PATH_SUFIX,
|
||||
status_code=HTTPStatus.CREATED,
|
||||
)
|
||||
else:
|
||||
requests_mock.register_uri(
|
||||
"POST",
|
||||
"http://127.0.0.1:8080" + SIGNAL_SEND_PATH_SUFIX,
|
||||
status_code=HTTPStatus.BAD_REQUEST,
|
||||
)
|
||||
if content_length_header is not None:
|
||||
requests_mock.register_uri(
|
||||
"GET",
|
||||
URL_ATTACHMENT,
|
||||
status_code=HTTPStatus.OK,
|
||||
content=CONTENT,
|
||||
headers={"Content-Length": content_length_header},
|
||||
)
|
||||
else:
|
||||
requests_mock.register_uri(
|
||||
"GET",
|
||||
URL_ATTACHMENT,
|
||||
status_code=HTTPStatus.OK,
|
||||
content=CONTENT,
|
||||
)
|
||||
return requests_mock
|
||||
|
||||
return _signal_requests_mock_factory
|
||||
|
||||
@ -1,23 +1,33 @@
|
||||
"""The tests for the signal_messenger platform."""
|
||||
import base64
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import tempfile
|
||||
from unittest.mock import patch
|
||||
|
||||
from pysignalclirestapi.api import SignalCliRestApiError
|
||||
import pytest
|
||||
from requests_mock.mocker import Mocker
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.components.signal_messenger.conftest import (
|
||||
CONTENT,
|
||||
MESSAGE,
|
||||
NUMBER_FROM,
|
||||
NUMBERS_TO,
|
||||
SIGNAL_SEND_PATH_SUFIX,
|
||||
URL_ATTACHMENT,
|
||||
SignalNotificationService,
|
||||
)
|
||||
|
||||
BASE_COMPONENT = "notify"
|
||||
|
||||
|
||||
async def test_signal_messenger_init(hass):
|
||||
async def test_signal_messenger_init(hass: HomeAssistant) -> None:
|
||||
"""Test that service loads successfully."""
|
||||
config = {
|
||||
BASE_COMPONENT: {
|
||||
@ -36,8 +46,13 @@ async def test_signal_messenger_init(hass):
|
||||
assert hass.services.has_service(BASE_COMPONENT, "test")
|
||||
|
||||
|
||||
def test_send_message(signal_notification_service, signal_requests_mock, caplog):
|
||||
def test_send_message(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
signal_requests_mock_factory: Mocker,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Test send message."""
|
||||
signal_requests_mock = signal_requests_mock_factory()
|
||||
with caplog.at_level(
|
||||
logging.DEBUG, logger="homeassistant.components.signal_messenger.notify"
|
||||
):
|
||||
@ -48,32 +63,58 @@ def test_send_message(signal_notification_service, signal_requests_mock, caplog)
|
||||
assert_sending_requests(signal_requests_mock)
|
||||
|
||||
|
||||
def test_send_message_should_show_deprecation_warning(
|
||||
signal_notification_service, signal_requests_mock, caplog
|
||||
):
|
||||
"""Test send message should show deprecation warning."""
|
||||
with caplog.at_level(
|
||||
logging.WARNING, logger="homeassistant.components.signal_messenger.notify"
|
||||
):
|
||||
send_message_with_attachment(signal_notification_service, True)
|
||||
|
||||
assert (
|
||||
"The 'attachment' option is deprecated, please replace it with 'attachments'. This option will become invalid in version 0.108"
|
||||
in caplog.text
|
||||
)
|
||||
assert signal_requests_mock.called
|
||||
assert signal_requests_mock.call_count == 2
|
||||
assert_sending_requests(signal_requests_mock, 1)
|
||||
|
||||
|
||||
def test_send_message_with_attachment(
|
||||
signal_notification_service, signal_requests_mock, caplog
|
||||
):
|
||||
"""Test send message with attachment."""
|
||||
def test_send_message_to_api_with_bad_data_throws_error(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
signal_requests_mock_factory: Mocker,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Test sending a message with bad data to the API throws an error."""
|
||||
signal_requests_mock = signal_requests_mock_factory(False)
|
||||
with caplog.at_level(
|
||||
logging.DEBUG, logger="homeassistant.components.signal_messenger.notify"
|
||||
):
|
||||
send_message_with_attachment(signal_notification_service, False)
|
||||
with pytest.raises(SignalCliRestApiError) as exc:
|
||||
signal_notification_service.send_message(MESSAGE)
|
||||
|
||||
assert "Sending signal message" in caplog.text
|
||||
assert signal_requests_mock.called
|
||||
assert signal_requests_mock.call_count == 2
|
||||
assert "Couldn't send signal message" in str(exc.value)
|
||||
|
||||
|
||||
def test_send_message_with_bad_data_throws_vol_error(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
signal_requests_mock_factory: Mocker,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Test sending a message with bad data throws an error."""
|
||||
with caplog.at_level(
|
||||
logging.DEBUG, logger="homeassistant.components.signal_messenger.notify"
|
||||
):
|
||||
with pytest.raises(vol.Invalid) as exc:
|
||||
data = {"test": "test"}
|
||||
signal_notification_service.send_message(MESSAGE, **{"data": data})
|
||||
|
||||
assert "Sending signal message" in caplog.text
|
||||
assert "extra keys not allowed" in str(exc.value)
|
||||
|
||||
|
||||
def test_send_message_with_attachment(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
signal_requests_mock_factory: Mocker,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Test send message with attachment."""
|
||||
signal_requests_mock = signal_requests_mock_factory()
|
||||
with caplog.at_level(
|
||||
logging.DEBUG, logger="homeassistant.components.signal_messenger.notify"
|
||||
):
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".png", prefix=os.path.basename(__file__)
|
||||
) as temp_file:
|
||||
temp_file.write("attachment_data")
|
||||
data = {"attachments": [temp_file.name]}
|
||||
signal_notification_service.send_message(MESSAGE, **{"data": data})
|
||||
|
||||
assert "Sending signal message" in caplog.text
|
||||
assert signal_requests_mock.called
|
||||
@ -81,19 +122,211 @@ def test_send_message_with_attachment(
|
||||
assert_sending_requests(signal_requests_mock, 1)
|
||||
|
||||
|
||||
def send_message_with_attachment(signal_notification_service, deprecated=False):
|
||||
"""Send message with attachment."""
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".png", prefix=os.path.basename(__file__)
|
||||
) as tf:
|
||||
tf.write("attachment_data")
|
||||
data = {"attachment": tf.name} if deprecated else {"attachments": [tf.name]}
|
||||
def test_send_message_with_attachment_as_url(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
signal_requests_mock_factory: Mocker,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Test send message with attachment as URL."""
|
||||
signal_requests_mock = signal_requests_mock_factory(True, str(len(CONTENT)))
|
||||
with caplog.at_level(
|
||||
logging.DEBUG, logger="homeassistant.components.signal_messenger.notify"
|
||||
):
|
||||
data = {"urls": [URL_ATTACHMENT]}
|
||||
signal_notification_service.send_message(MESSAGE, **{"data": data})
|
||||
|
||||
assert "Sending signal message" in caplog.text
|
||||
assert signal_requests_mock.called
|
||||
assert signal_requests_mock.call_count == 3
|
||||
assert_sending_requests(signal_requests_mock, 1)
|
||||
|
||||
def assert_sending_requests(signal_requests_mock, attachments_num=0):
|
||||
|
||||
def test_get_attachments(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
signal_requests_mock_factory: Mocker,
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
"""Test getting attachments as URL."""
|
||||
signal_requests_mock = signal_requests_mock_factory(True, str(len(CONTENT)))
|
||||
data = {"urls": [URL_ATTACHMENT]}
|
||||
result = signal_notification_service.get_attachments_as_bytes(
|
||||
data, len(CONTENT), hass
|
||||
)
|
||||
|
||||
assert signal_requests_mock.called
|
||||
assert signal_requests_mock.call_count == 1
|
||||
assert result == [bytearray(CONTENT)]
|
||||
|
||||
|
||||
def test_get_attachments_not_on_allowlist(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
"""Test getting attachments as URL that aren't on the allowlist."""
|
||||
url = "http://dodgyurl.com"
|
||||
data = {"urls": [url]}
|
||||
with caplog.at_level(
|
||||
logging.ERROR, logger="homeassistant.components.signal_messenger.notify"
|
||||
):
|
||||
result = signal_notification_service.get_attachments_as_bytes(
|
||||
data, len(CONTENT), hass
|
||||
)
|
||||
|
||||
assert f"URL '{url}' not in allow list" in caplog.text
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_get_attachments_with_large_attachment(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
signal_requests_mock_factory: Mocker,
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
"""Test getting attachments as URL with large attachment (per Content-Length header) throws error."""
|
||||
signal_requests_mock = signal_requests_mock_factory(True, str(len(CONTENT) + 1))
|
||||
with pytest.raises(ValueError) as exc:
|
||||
data = {"urls": [URL_ATTACHMENT]}
|
||||
signal_notification_service.get_attachments_as_bytes(data, len(CONTENT), hass)
|
||||
|
||||
assert signal_requests_mock.called
|
||||
assert signal_requests_mock.call_count == 1
|
||||
assert "Attachment too large (Content-Length reports" in str(exc.value)
|
||||
|
||||
|
||||
def test_get_attachments_with_large_attachment_no_header(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
signal_requests_mock_factory: Mocker,
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
"""Test getting attachments as URL with large attachment (per content length) throws error."""
|
||||
signal_requests_mock = signal_requests_mock_factory()
|
||||
with pytest.raises(ValueError) as exc:
|
||||
data = {"urls": [URL_ATTACHMENT]}
|
||||
signal_notification_service.get_attachments_as_bytes(
|
||||
data, len(CONTENT) - 1, hass
|
||||
)
|
||||
|
||||
assert signal_requests_mock.called
|
||||
assert signal_requests_mock.call_count == 1
|
||||
assert "Attachment too large (Stream reports" in str(exc.value)
|
||||
|
||||
|
||||
def test_get_filenames_with_none_data(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
) -> None:
|
||||
"""Test getting filenames with None data returns None."""
|
||||
data = None
|
||||
result = signal_notification_service.get_filenames(data)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_get_filenames_with_attachments_data(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
) -> None:
|
||||
"""Test getting filenames with 'attachments' in data."""
|
||||
data = {"attachments": ["test"]}
|
||||
result = signal_notification_service.get_filenames(data)
|
||||
|
||||
assert result == ["test"]
|
||||
|
||||
|
||||
def test_get_filenames_with_multiple_attachments_data(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
) -> None:
|
||||
"""Test getting filenames with multiple 'attachments' in data."""
|
||||
data = {"attachments": ["test", "test2"]}
|
||||
result = signal_notification_service.get_filenames(data)
|
||||
|
||||
assert result == ["test", "test2"]
|
||||
|
||||
|
||||
def test_get_filenames_with_non_list_returns_none(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
) -> None:
|
||||
"""Test getting filenames with non list data."""
|
||||
data = {"attachments": "test"}
|
||||
result = signal_notification_service.get_filenames(data)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_get_attachments_with_non_list_returns_none(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
"""Test getting attachments with non list data."""
|
||||
data = {"urls": URL_ATTACHMENT}
|
||||
result = signal_notification_service.get_attachments_as_bytes(
|
||||
data, len(CONTENT), hass
|
||||
)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_get_attachments_with_verify_unset(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
signal_requests_mock_factory: Mocker,
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
"""Test getting attachments as URL with verify_ssl unset results in verify=true."""
|
||||
signal_requests_mock = signal_requests_mock_factory()
|
||||
data = {"urls": [URL_ATTACHMENT]}
|
||||
signal_notification_service.get_attachments_as_bytes(data, len(CONTENT), hass)
|
||||
|
||||
assert signal_requests_mock.called
|
||||
assert signal_requests_mock.call_count == 1
|
||||
assert signal_requests_mock.last_request.verify is True
|
||||
|
||||
|
||||
def test_get_attachments_with_verify_set_true(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
signal_requests_mock_factory: Mocker,
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
"""Test getting attachments as URL with verify_ssl set to true results in verify=true."""
|
||||
signal_requests_mock = signal_requests_mock_factory()
|
||||
data = {"verify_ssl": True, "urls": [URL_ATTACHMENT]}
|
||||
signal_notification_service.get_attachments_as_bytes(data, len(CONTENT), hass)
|
||||
|
||||
assert signal_requests_mock.called
|
||||
assert signal_requests_mock.call_count == 1
|
||||
assert signal_requests_mock.last_request.verify is True
|
||||
|
||||
|
||||
def test_get_attachments_with_verify_set_false(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
signal_requests_mock_factory: Mocker,
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
"""Test getting attachments as URL with verify_ssl set to false results in verify=false."""
|
||||
signal_requests_mock = signal_requests_mock_factory()
|
||||
data = {"verify_ssl": False, "urls": [URL_ATTACHMENT]}
|
||||
signal_notification_service.get_attachments_as_bytes(data, len(CONTENT), hass)
|
||||
|
||||
assert signal_requests_mock.called
|
||||
assert signal_requests_mock.call_count == 1
|
||||
assert signal_requests_mock.last_request.verify is False
|
||||
|
||||
|
||||
def test_get_attachments_with_verify_set_garbage(
|
||||
signal_notification_service: SignalNotificationService,
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
"""Test getting attachments as URL with verify_ssl set to garbage results in None."""
|
||||
data = {"verify_ssl": "test", "urls": [URL_ATTACHMENT]}
|
||||
result = signal_notification_service.get_attachments_as_bytes(
|
||||
data, len(CONTENT), hass
|
||||
)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
def assert_sending_requests(
|
||||
signal_requests_mock_factory: Mocker, attachments_num: int = 0
|
||||
) -> None:
|
||||
"""Assert message was send with correct parameters."""
|
||||
send_request = signal_requests_mock.request_history[-1]
|
||||
send_request = signal_requests_mock_factory.request_history[-1]
|
||||
assert send_request.path == SIGNAL_SEND_PATH_SUFIX
|
||||
|
||||
body_request = json.loads(send_request.text)
|
||||
@ -101,3 +334,7 @@ def assert_sending_requests(signal_requests_mock, attachments_num=0):
|
||||
assert body_request["number"] == NUMBER_FROM
|
||||
assert body_request["recipients"] == NUMBERS_TO
|
||||
assert len(body_request["base64_attachments"]) == attachments_num
|
||||
|
||||
for attachment in body_request["base64_attachments"]:
|
||||
if len(attachment) > 0:
|
||||
assert base64.b64decode(attachment) == CONTENT
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user