github
/
apprise
mirror of https://github.com/caronc/apprise
1
0
Fork 0
Code Issues Releases Wiki Activity

updated test coverage + code cleanup

pull/1374/head
Chris Caron 2025-08-01 15:58:00 -04:00
parent 6dc3324dd1
commit e3ce566a57
2 changed files with 88 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

81
apprise/plugins/signl4.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
@ -31,14 +30,16 @@
#
from json import dumps
from typing import Any, Optional
import requests
from .base import NotifyBase
from ..url import PrivacyMode
from ..common import NotifyType
from ..utils.parse import validate_regex
from ..locale import gettext_lazy as _
from ..url import PrivacyMode
from ..utils.parse import parse_bool, validate_regex
from .base import NotifyBase
class NotifySIGNL4(NotifyBase):
"""
@ -49,7 +50,7 @@ class NotifySIGNL4(NotifyBase):
service_name = "SIGNL4"
# The services URL
service_url = "https://connect.signl4.com/webhook/"
service_url = "https://signl4.com/"
# Secure Protocol
secure_protocol = "signl4"
@ -60,6 +61,9 @@ class NotifySIGNL4(NotifyBase):
# Our event action type
event_action = "trigger"
# Our default notification URL
notify_url = "https://connect.signl4.com/webhook/{secret}/"
# Define object templates
templates = (
"{schema}://{secret}",
@ -79,40 +83,43 @@ class NotifySIGNL4(NotifyBase):
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
"service": {
"name": _("service"),
"name": _("Service"),
"type": "string",
},
"location": {
"name": _("location"),
"name": _("Location"),
"type": "string",
},
"alerting_scenario": {
"name": _("alerting_scenario"),
"name": _("Alerting Scenario"),
"type": "string",
},
"filtering": {
"name": _("filtering"),
"name": _("Filtering"),
"type": "bool",
"default": False,
},
"external_id": {
"name": _("external_id"),
"name": _("External ID"),
"type": "string",
},
"status": {
"name": _("status"),
"name": _("Status"),
"type": "string",
},
})
def __init__(self,
secret,
service=None,
location=None,
alerting_scenario=None,
filtering=None,
external_id=None,
status=None,
**kwargs):
def __init__(
self,
secret: str,
service: Optional[str] = None,
location: Optional[str] = None,
alerting_scenario: Optional[str] = None,
filtering: Optional[bool] = None,
external_id: Optional[str] = None,
status: Optional[str] = None,
**kwargs: Any
) -> None:
"""
Initialize SIGNL4 Object
"""
@ -136,7 +143,11 @@ class NotifySIGNL4(NotifyBase):
self.alerting_scenario = alerting_scenario
# A filtering option for notifications
self.filtering = bool(filtering)
self.filtering = (
self.template_args["filtering"]["default"]
if filtering is None
else bool(filtering)
)
# A external_id option for notifications
self.external_id = external_id
@ -160,7 +171,7 @@ class NotifySIGNL4(NotifyBase):
payload = {
"title": title if title else self.app_desc,
"body": body,
"X-S4-SourceSystem": "Apprise",
"X-S4-SourceSystem": self.app_id,
}
if self.service:
@ -182,11 +193,13 @@ class NotifySIGNL4(NotifyBase):
payload["X-S4-Status"] = self.status
# Prepare our URL
notify_url = self.service_url + self.secret
notify_url = self.notify_url.format(secret=self.secret)
self.logger.debug(
"SIGNL4 POST URL: %s (cert_verify=%s)",
notify_url, self.verify_certificate)
self.logger.debug("SIGNL4 Payload: %r", payload)
self.logger.debug("SIGNL4 POST URL: {notify_url} \
(cert_verify={self.verify_certificate}")
self.logger.debug("SIGNL4 Payload: {payload}")
# Always call throttle before any remote server i/o is made
self.throttle()
@ -214,7 +227,7 @@ class NotifySIGNL4(NotifyBase):
", " if status_str else "",
r.status_code))
self.logger.debug("Response Details:\r\n{}".format(r.content))
self.logger.debug("Response Details:\r\n%r", r.content)
# Return; we're done
return False
@ -225,8 +238,8 @@ class NotifySIGNL4(NotifyBase):
except requests.RequestException as e:
self.logger.warning(
"A Connection error occurred sending SIGNL4 "
"notification to {self.host}.")
self.logger.debug("Socket Exception: " + str(e))
"notification to %s", self.host)
self.logger.debug("Socket Exception: %s", str(e))
# Return; we're done
return False
@ -261,8 +274,9 @@ class NotifySIGNL4(NotifyBase):
if self.alerting_scenario is not None:
params["alerting_scenario"] = self.alerting_scenario
if self.filtering is not None:
params["filtering"] = self.filtering
if self.filtering != self.template_args["filtering"]["default"]:
# Only add filtering if it is not the default value
params["filtering"] = "yes" if self.filtering else "no"
if self.external_id is not None:
params["external_id"] = self.external_id
@ -319,7 +333,10 @@ class NotifySIGNL4(NotifyBase):
if "filtering" in results["qsd"] and len(results["qsd"]["filtering"]):
results["filtering"] = \
NotifySIGNL4.unquote(results["qsd"]["filtering"])
parse_bool(
NotifySIGNL4.unquote(
results["qsd"]["filtering"],
NotifySIGNL4.template_args["filtering"]["default"]))
if "external_id" in results["qsd"] and \
len(results["qsd"]["external_id"]):

65
tests/test_plugin_signl4.py
View File

@ -24,31 +24,6 @@
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2025, Chris Caron <lead2gold@gmail.com>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
from json import dumps
@ -101,6 +76,16 @@ apprise_url_tests = (
"requests_response_text": SIGNL4_GOOD_RESPONSE,
},
),
(
"signl4://?secret=secret",
{
# No targets specified; this is allowed
"instance": NotifySIGNL4,
"notify_type": NotifyType.FAILURE,
# Our response expected server response
"requests_response_text": SIGNL4_GOOD_RESPONSE,
},
),
(
"signl4://secret/?service=IoT",
{
@ -111,7 +96,17 @@ apprise_url_tests = (
"requests_response_text": SIGNL4_GOOD_RESPONSE,
},
),
(
(
"signl4://secret/?filtering=yes",
{
# European Region
"instance": NotifySIGNL4,
"notify_type": NotifyType.FAILURE,
# Our response expected server response
"requests_response_text": SIGNL4_GOOD_RESPONSE,
},
),
(
"signl4://secret/?location=40.6413111,-73.7781391",
{
# European Region
@ -151,6 +146,24 @@ apprise_url_tests = (
"requests_response_text": SIGNL4_GOOD_RESPONSE,
},
),
(
"signl4://secret/",
{
"instance": NotifySIGNL4,
# throw a bizzare code forcing us to fail to look it up
"response": False,
"requests_response_code": 999,
},
),
(
"signl4://secret/",
{
"instance": NotifySIGNL4,
# Throws a series of i/o exceptions with this flag
# is set and tests that we gracfully handle them
"test_requests_exceptions": True,
},
),
)
def test_plugin_signl4_urls():