From 31100ad4bb495181065dcf393891c58d2964f446 Mon Sep 17 00:00:00 2001 From: Andreas Motl Date: Sun, 12 Mar 2023 18:33:42 +0100 Subject: [PATCH] [Pushsafer] Implement v2 configuration layout, where `addrs` is a dict --- mqttwarn/model.py | 8 +- mqttwarn/services/pushsafer.py | 84 ++++++++++- tests/services/pushsafer/test_pushsafer_v2.py | 137 ++++++++++++++++++ 3 files changed, 227 insertions(+), 2 deletions(-) create mode 100644 tests/services/pushsafer/test_pushsafer_v2.py diff --git a/mqttwarn/model.py b/mqttwarn/model.py index bed32ba0..f2455b10 100644 --- a/mqttwarn/model.py +++ b/mqttwarn/model.py @@ -37,6 +37,12 @@ def enum(self): return item +# Covering old- and new-style configuration layouts. `addrs` has +# originally been a list of strings, has been expanded to be a +# list of dictionaries (Apprise), and to be a dictionary (Pushsafer). +addrs_type = Union[List[Union[str, Dict[str, str]]], Dict[str, str]] + + @dataclass class ProcessorItem: """ @@ -46,7 +52,7 @@ class ProcessorItem: service: Optional[str] = None target: Optional[str] = None config: Dict = field(default_factory=dict) - addrs: List[Union[str, Dict[str, str]]] = field(default_factory=list) + addrs: addrs_type = field(default_factory=list) # type: ignore[assignment] priority: Optional[int] = None topic: Optional[str] = None title: Optional[str] = None diff --git a/mqttwarn/services/pushsafer.py b/mqttwarn/services/pushsafer.py index a6bd329d..9bfd70b0 100644 --- a/mqttwarn/services/pushsafer.py +++ b/mqttwarn/services/pushsafer.py @@ -5,6 +5,9 @@ __copyright__ = 'Copyright 2014 Jan-Piet Mens' __license__ = 'Eclipse Public License - v 1.0 (http://www.eclipse.org/legal/epl-v10.html)' +import dataclasses +from collections import OrderedDict + from future import standard_library standard_library.install_aliases() import os @@ -83,7 +86,7 @@ def encode(self): if isinstance(addrs, t.List): self.encode_v1() elif isinstance(addrs, t.Dict): - raise NotImplementedError("Pushsafer configuration layout v2 not implemented yet") + self.encode_v2() else: raise ValueError(f"Unable to decode Pushsafer configuration layout. type={type(addrs)}") @@ -156,3 +159,82 @@ def encode_v1(self): params['t'] = title self.params = params + + def encode_v2(self): + """ + New-style configuration layout with named parameters for Pushsafer. + """ + + addrs = self.item.addrs + title = self.item.title + + # Decode Private or Alias Key. + try: + self.private_key = addrs["private_key"] + except KeyError: + raise PushsaferConfigurationError(f"Pushsafer private or alias key not configured") + + params = { + 'expire': 3600, + } + + # Decode and serialize all other parameters. + pp = PushsaferParameters(**addrs) + params.update(pp.translated()) + + # Propagate `title` separately. + if title is not None: + params['t'] = title + + self.params = params + + +@dataclasses.dataclass +class PushsaferParameters: + """ + Manage available Pushsafer parameters, and map them to their short representations, + suitable for sending over the wire. + + - https://www.pushsafer.com/en/pushapi + - https://www.pushsafer.com/en/pushapi_ext + """ + private_key: t.Optional[str] = None + device: t.Optional[str] = None + icon: t.Optional[int] = None + sound: t.Optional[int] = None + vibration: t.Optional[int] = None + url: t.Optional[str] = None + url_title: t.Optional[str] = None + time_to_live: t.Optional[int] = None + priority: t.Optional[int] = None + retry: t.Optional[int] = None + expire: t.Optional[int] = None + answer: t.Optional[int] = None + + PARAMETER_MAP = { + "device": "d", + "icon": "i", + "sound": "s", + "vibration": "v", + "url": "u", + "url_title": "ut", + "time_to_live": "l", + "priority": "pr", + "retry": "re", + "expire": "ex", + "answer": "a", + } + + def to_dict(self) -> t.Dict[str, t.Union[str, int]]: + return dataclasses.asdict(self) + + def translated(self) -> t.Dict[str, t.Union[str, int]]: + """ + Translate parameters to their wire representations. + """ + result = OrderedDict() + for attribute, parameter_name in self.PARAMETER_MAP.items(): + value = getattr(self, attribute) + if value is not None: + result[parameter_name] = value + return result diff --git a/tests/services/pushsafer/test_pushsafer_v2.py b/tests/services/pushsafer/test_pushsafer_v2.py new file mode 100644 index 00000000..389d2db3 --- /dev/null +++ b/tests/services/pushsafer/test_pushsafer_v2.py @@ -0,0 +1,137 @@ +# -*- coding: utf-8 -*- +# (c) 2023 The mqttwarn developers +""" +This file contains test cases for the v2 configuration layout variant, +where `addrs` is a dictionary. +""" +import dataclasses +import re +import typing as t +from operator import attrgetter + +import pytest + +from mqttwarn.model import ProcessorItem as Item +from mqttwarn.util import load_module_from_file +from tests.services.pushsafer.util import TEST_TOKEN, assert_request, get_reference_data + + +def test_pushsafer_parameter_failure(srv, caplog, mock_urlopen_success): + """ + Test Pushsafer service with missing `private_key` parameter. It should fail. + """ + + module = load_module_from_file("mqttwarn/services/pushsafer.py") + item = Item(addrs={}, message="⚽ Notification message ⚽") + outcome = module.plugin(srv, item) + + assert outcome is False + assert "Pushsafer private or alias key not configured. target=None" in caplog.messages + + +def test_pushsafer_basic_success(srv, caplog, mock_urlopen_success): + """ + Test Pushsafer service with only `private_key` parameter. + """ + + module = load_module_from_file("mqttwarn/services/pushsafer.py") + item = Item(addrs={"private_key": TEST_TOKEN}, message="⚽ Notification message ⚽") + outcome = module.plugin(srv, item) + + request = mock_urlopen_success.call_args[0][0] + assert_request(request, get_reference_data()) + + assert outcome is True + assert "Sending pushsafer notification" in caplog.text + assert "Successfully sent pushsafer notification" in caplog.text + + +def test_pushsafer_basic_failure(srv, caplog, mock_urlopen_failure): + """ + Test Pushsafer service with a response indicating an error. + """ + + module = load_module_from_file("mqttwarn/services/pushsafer.py") + item = Item(addrs={"private_key": TEST_TOKEN}, message="⚽ Notification message ⚽") + outcome = module.plugin(srv, item) + + assert outcome is False + assert "Sending pushsafer notification" in caplog.text + assert re.match('.*Error sending pushsafer notification.*{"status": 6}.*', caplog.text, re.DOTALL) + + +def test_pushsafer_title_success(srv, caplog, mock_urlopen_success): + """ + Test Pushsafer service with title. + """ + + module = load_module_from_file("mqttwarn/services/pushsafer.py") + item = Item(addrs={"private_key": TEST_TOKEN}, message="⚽ Notification message ⚽", title="⚽ Message title ⚽") + outcome = module.plugin(srv, item) + + request = mock_urlopen_success.call_args[0][0] + assert_request(request, get_reference_data(t="⚽ Message title ⚽")) + + assert outcome is True + assert "Sending pushsafer notification" in caplog.text + assert "Successfully sent pushsafer notification" in caplog.text + + +def test_pushsafer_token_environment_success(srv, caplog, mocker, mock_urlopen_success): + """ + Test Pushsafer service with token from `PUSHSAFER_TOKEN` environment variable. + """ + mocker.patch.dict("os.environ", {"PUSHSAFER_TOKEN": TEST_TOKEN}) + + module = load_module_from_file("mqttwarn/services/pushsafer.py") + item = Item(addrs={"private_key": ""}, message="⚽ Notification message ⚽") + outcome = module.plugin(srv, item) + + request = mock_urlopen_success.call_args[0][0] + assert_request(request, get_reference_data()) + + assert outcome is True + assert "Sending pushsafer notification" in caplog.text + assert "Successfully sent pushsafer notification" in caplog.text + + +@dataclasses.dataclass +class IoTestItem: + id: str # noqa: A003 + in_addrs: t.Dict[str, str] + out_data: t.Dict[str, str] + + +variants = [ + IoTestItem(id="device-id", in_addrs={"private_key": TEST_TOKEN, "device": "52|65|78"}, out_data={"d": "52|65|78"}), + IoTestItem(id="icon", in_addrs={"private_key": TEST_TOKEN, "icon": "test.ico"}, out_data={"i": "test.ico"}), + IoTestItem(id="sound", in_addrs={"private_key": TEST_TOKEN, "sound": "test.mp3"}, out_data={"s": "test.mp3"}), + IoTestItem(id="vibration", in_addrs={"private_key": TEST_TOKEN, "vibration": 1}, out_data={"v": "1"}), + IoTestItem( + id="url", + in_addrs={"private_key": TEST_TOKEN, "url": "http://example.org"}, + out_data={"u": "http://example.org"}, + ), + IoTestItem( + id="url-title", in_addrs={"private_key": TEST_TOKEN, "url_title": "Example Org"}, out_data={"ut": "Example Org"} + ), + IoTestItem(id="time-to-live", in_addrs={"private_key": TEST_TOKEN, "time_to_live": 60}, out_data={"l": "60"}), + IoTestItem(id="priority", in_addrs={"private_key": TEST_TOKEN, "priority": 2}, out_data={"pr": "2"}), + IoTestItem(id="retry", in_addrs={"private_key": TEST_TOKEN, "retry": 60}, out_data={"re": "60"}), + IoTestItem(id="expire", in_addrs={"private_key": TEST_TOKEN, "expire": 600}, out_data={"ex": "600"}), + IoTestItem(id="answer", in_addrs={"private_key": TEST_TOKEN, "answer": 1}, out_data={"a": "1"}), +] + + +@pytest.mark.parametrize("variant", variants, ids=attrgetter("id")) +def test_pushsafer_variant(srv, caplog, mock_urlopen_success, variant: IoTestItem): + module = load_module_from_file("mqttwarn/services/pushsafer.py") + item = Item(addrs=variant.in_addrs, message="⚽ Notification message ⚽") + outcome = module.plugin(srv, item) + + request = mock_urlopen_success.call_args[0][0] + assert_request(request, get_reference_data(**variant.out_data)) + + assert outcome is True + assert "Sending pushsafer notification" in caplog.text + assert "Successfully sent pushsafer notification" in caplog.text