DGNum/uptime-kuma-api
6
0
Fork 0
Code Issues 17 Pull requests 5 Projects Releases 24 Packages Wiki Activity Actions

feat: add support for uptime kuma 1.22.0 and 1.22.1

Browse source
This commit is contained in:
lucasheld 2023-07-07 22:28:20 +02:00
parent 934ab15457
commit 06f1173569
8 changed files with 136 additions and 61 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
docs/api.rst
View file

@ -19,6 +19,9 @@ Enums
.. autoclass:: MonitorType
:members:
.. autoclass:: MonitorStatus
:members:
.. autoclass:: NotificationType
:members:

2
run_tests.sh
View file

@ -5,7 +5,7 @@ if [ $version ]
then
versions=("$version")
else
versions=(1.21.3)
versions=(1.22.1 1.22.0 1.21.3)
fi
for version in ${versions[*]}

6
scripts/build_notification_docstring.py
View file

@ -89,6 +89,12 @@ data = {
"twilioToNumber": "To Number.",
"twilioFromNumber": "From Number.",
"pushoverttl": "Message TTL (Seconds).",
"ntfyaccesstoken": "Access Token.",
"ntfyAuthenticationMethod": "Authentication Method.",
}
for provider in notification_provider_options:

21
tests/test_monitor.py
View file

@ -1,4 +1,5 @@
import unittest
from packaging.version import parse as parse_version
from uptime_kuma_api import UptimeKumaException, MonitorType, AuthMethod, MonitorStatus
from uptime_kuma_test_case import UptimeKumaTestCase
@ -303,6 +304,26 @@ class TestMonitor(UptimeKumaTestCase):
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_group(self):
if parse_version(self.api.version) < parse_version("1.22"):
self.skipTest("Unsupported in this Uptime Kuma version")
# create monitor group
expected_monitor = {
"type": MonitorType.GROUP,
"name": "monitor 1"
}
group_monitor = self.do_test_monitor_type(expected_monitor)
group_monitor_id = group_monitor["id"]
# use monitor group as parent for another monitor
expected_monitor = {
"type": MonitorType.PUSH,
"name": "monitor 1",
"parent": group_monitor_id
}
self.do_test_monitor_type(expected_monitor)
def test_delete_not_existing_monitor(self):
with self.assertRaises(UptimeKumaException):
self.api.delete_monitor(42)

146
uptime_kuma_api/api.py
View file

@ -11,6 +11,7 @@ from typing import Any
import requests
import socketio
from packaging.version import parse as parse_version
from . import (
AuthMethod,
@ -58,7 +59,11 @@ def parse_value(data, key, type_, default=None) -> None:
else:
if key in data:
if data[key] is not None:
try:
data[key] = type_(data[key])
except ValueError:
# todo: add warning to logs
pass
elif default is not None:
data[key] = default
@ -128,10 +133,10 @@ def _convert_monitor_input(kwargs) -> None:
kwargs["databaseConnectionString"] = "postgres://username:password@host:port/database"
elif kwargs["type"] == MonitorType.MYSQL:
kwargs["databaseConnectionString"] = "mysql://username:password@host:port/database"
elif kwargs["type"] == MonitorType.MONGODB:
kwargs["databaseConnectionString"] = "mongodb://username:password@host:port/database"
elif kwargs["type"] == MonitorType.REDIS:
kwargs["databaseConnectionString"] = "redis://user:password@host:port"
elif kwargs["type"] == MonitorType.MONGODB:
kwargs["databaseConnectionString"] = "mongodb://username:password@host:port/database"
if kwargs["type"] == MonitorType.PUSH and not kwargs.get("pushToken"):
kwargs["pushToken"] = gen_secret(10)
@ -187,49 +192,6 @@ def _build_proxy_data(
return data
def _build_status_page_data(
slug: str,
# config
id: int,
title: str,
description: str = None,
theme: str = "light",
published: bool = True,
showTags: bool = False,
domainNameList: list = None,
googleAnalyticsId: str = None,
customCSS: str = "",
footerText: str = None,
showPoweredBy: bool = True,
icon: str = "/icon.svg",
publicGroupList: list = None
) -> tuple[str, dict, str, list]:
if theme not in ["light", "dark"]:
raise ValueError
if not domainNameList:
domainNameList = []
if not publicGroupList:
publicGroupList = []
config = {
"id": id,
"slug": slug,
"title": title,
"description": description,
"icon": icon,
"theme": theme,
"published": published,
"showTags": showTags,
"domainNameList": domainNameList,
"googleAnalyticsId": googleAnalyticsId,
"customCSS": customCSS,
"footerText": footerText,
"showPoweredBy": showPoweredBy
}
return slug, config, icon, publicGroupList
def _convert_docker_host_input(kwargs) -> None:
if not kwargs["dockerDaemon"]:
if kwargs["dockerType"] == DockerType.SOCKET:
@ -315,7 +277,8 @@ def _check_arguments_monitor(kwargs) -> None:
MonitorType.MYSQL: [],
MonitorType.MONGODB: [],
MonitorType.RADIUS: ["radiusUsername", "radiusPassword", "radiusSecret", "radiusCalledStationId", "radiusCallingStationId"],
MonitorType.REDIS: []
MonitorType.REDIS: [],
MonitorType.GROUP: []
}
type_ = kwargs["type"]
required_args = required_args_by_type[type_]
@ -678,6 +641,7 @@ class UptimeKumaApi(object):
self,
type: MonitorType,
name: str,
parent: int = None,
description: str = None,
interval: int = 60,
retryInterval: int = 60,
@ -770,6 +734,11 @@ class UptimeKumaApi(object):
"httpBodyEncoding": httpBodyEncoding
}
if parse_version(self.version) >= parse_version("1.22"):
data.update({
"parent": parent
})
if type in [MonitorType.KEYWORD, MonitorType.GRPC_KEYWORD]:
data.update({
"keyword": keyword,
@ -939,6 +908,54 @@ class UptimeKumaApi(object):
}
return data
def _build_status_page_data(
self,
slug: str,
# config
id: int,
title: str,
description: str = None,
theme: str = None,
published: bool = True,
showTags: bool = False,
domainNameList: list = None,
googleAnalyticsId: str = None,
customCSS: str = "",
footerText: str = None,
showPoweredBy: bool = True,
icon: str = "/icon.svg",
publicGroupList: list = None
) -> tuple[str, dict, str, list]:
if not theme:
if parse_version(self.version) >= parse_version("1.22"):
theme = "auto"
else:
theme = "light"
if theme not in ["auto", "light", "dark"]:
raise ValueError
if not domainNameList:
domainNameList = []
if not publicGroupList:
publicGroupList = []
config = {
"id": id,
"slug": slug,
"title": title,
"description": description,
"icon": icon,
"theme": theme,
"published": published,
"showTags": showTags,
"domainNameList": domainNameList,
"googleAnalyticsId": googleAnalyticsId,
"customCSS": customCSS,
"footerText": footerText,
"showPoweredBy": showPoweredBy
}
return slug, config, icon, publicGroupList
# monitor
def get_monitors(self) -> list[dict]:
@ -961,14 +978,17 @@ class UptimeKumaApi(object):
'basic_auth_pass': None,
'basic_auth_user': None,
'body': None,
'childrenIDs': [],
'databaseConnectionString': None,
'databaseQuery': None,
'description': None,
'dns_last_result': None,
'dns_resolve_server': '1.1.1.1',
'dns_resolve_type': 'A',
'docker_container': None,
'docker_host': None,
'expiryNotification': False,
'forceInactive': False,
'game': None,
'grpcBody': None,
'grpcEnableTls': False,
@ -979,6 +999,7 @@ class UptimeKumaApi(object):
'grpcUrl': None,
'headers': None,
'hostname': None,
'httpBodyEncoding': 'json',
'id': 1,
'ignoreTls': False,
'includeSensitiveData': True,
@ -986,7 +1007,7 @@ class UptimeKumaApi(object):
'keyword': None,
'maintenance': False,
'maxredirects': 10,
'maxretries': 1,
'maxretries': 0,
'method': 'GET',
'mqttPassword': None,
'mqttSuccessMessage': None,
@ -995,6 +1016,8 @@ class UptimeKumaApi(object):
'name': 'monitor 1',
'notificationIDList': [1, 2],
'packetSize': 56,
'parent': None,
'pathName': 'monitor 1',
'port': None,
'proxyId': None,
'pushToken': None,
@ -1006,7 +1029,10 @@ class UptimeKumaApi(object):
'resendInterval': 0,
'retryInterval': 60,
'tags': [],
'type': <MonitorType.HTTP: 'http'>
'tlsCa': None,
'tlsCert': None,
'tlsKey': None,
'type': <MonitorType.HTTP: 'http'>,
'upsideDown': False,
'url': 'http://127.0.0.1',
'weight': 2000
@ -1045,14 +1071,17 @@ class UptimeKumaApi(object):
'basic_auth_pass': None,
'basic_auth_user': None,
'body': None,
'childrenIDs': [],
'databaseConnectionString': None,
'databaseQuery': None,
'description': None,
'dns_last_result': None,
'dns_resolve_server': '1.1.1.1',
'dns_resolve_type': 'A',
'docker_container': None,
'docker_host': None,
'expiryNotification': False,
'forceInactive': False,
'game': None,
'grpcBody': None,
'grpcEnableTls': False,
@ -1063,6 +1092,7 @@ class UptimeKumaApi(object):
'grpcUrl': None,
'headers': None,
'hostname': None,
'httpBodyEncoding': 'json',
'id': 1,
'ignoreTls': False,
'includeSensitiveData': True,
@ -1070,7 +1100,7 @@ class UptimeKumaApi(object):
'keyword': None,
'maintenance': False,
'maxredirects': 10,
'maxretries': 1,
'maxretries': 0,
'method': 'GET',
'mqttPassword': None,
'mqttSuccessMessage': None,
@ -1079,6 +1109,8 @@ class UptimeKumaApi(object):
'name': 'monitor 1',
'notificationIDList': [1, 2],
'packetSize': 56,
'parent': None,
'pathName': 'monitor 1',
'port': None,
'proxyId': None,
'pushToken': None,
@ -1090,7 +1122,10 @@ class UptimeKumaApi(object):
'resendInterval': 0,
'retryInterval': 60,
'tags': [],
'type': <MonitorType.HTTP: 'http'>
'tlsCa': None,
'tlsCert': None,
'tlsKey': None,
'type': <MonitorType.HTTP: 'http'>,
'upsideDown': False,
'url': 'http://127.0.0.1',
'weight': 2000
@ -1781,7 +1816,6 @@ class UptimeKumaApi(object):
'monitorList': [
{
'id': 1,
'maintenance': False,
'name': 'monitor 1',
'sendUrl': 0
}
@ -1871,7 +1905,7 @@ class UptimeKumaApi(object):
:param int id: Id of the status page to save
:param str title: Title
:param str, optional description: Description, defaults to None
:param str, optional theme: Switch Theme, defaults to "light"
:param str, optional theme: Switch Theme, defaults to "auto"
:param bool, optional published: Published, defaults to True
:param bool, optional showTags: Show Tags, defaults to False
:param list, optional domainNameList: Domain Names, defaults to None
@ -1923,7 +1957,7 @@ class UptimeKumaApi(object):
status_page.pop("incident")
status_page.pop("maintenanceList")
status_page.update(kwargs)
data = _build_status_page_data(**status_page)
data = self._build_status_page_data(**status_page)
r = self._call('saveStatusPage', data)
# uptime kuma does not send the status page list event when a status page is saved
@ -3460,12 +3494,10 @@ class UptimeKumaApi(object):
>>> api.get_monitor_maintenance(1)
[
{
"id": 1,
"name": "monitor 1"
"id": 1
},
{
"id": 2,
"name": "monitor 2"
"id": 2
}
]
"""

6
uptime_kuma_api/docstrings.py
View file

@ -18,6 +18,7 @@ def monitor_docstring(mode) -> str:
return f"""
:param MonitorType{", optional" if mode == "edit" else ""} type: Monitor Type
:param str{", optional" if mode == "edit" else ""} name: Friendly Name
:param str, optional parent: Id of the monitor group, defaults to None
:param str, optional description: Description, defaults to None
:param int, optional interval: Heartbeat Interval, defaults to 60
:param int, optional retryInterval: Retry every X seconds, defaults to 60
@ -133,12 +134,14 @@ def notification_docstring(mode) -> str:
:param str mattermostchannel: Notification option for ``type`` :attr:`~.NotificationType.MATTERMOST`.
:param str mattermosticonemo: Notification option for ``type`` :attr:`~.NotificationType.MATTERMOST`.
:param str mattermosticonurl: Notification option for ``type`` :attr:`~.NotificationType.MATTERMOST`.
:param str ntfyAuthenticationMethod: Notification option for ``type`` :attr:`~.NotificationType.NTFY`. Authentication Method.
:param str ntfyusername: Notification option for ``type`` :attr:`~.NotificationType.NTFY`.
:param str ntfypassword: Notification option for ``type`` :attr:`~.NotificationType.NTFY`.
:param str ntfyaccesstoken: Notification option for ``type`` :attr:`~.NotificationType.NTFY`. Access Token.
:param str, optional ntfytopic: Notification option for ``type`` :attr:`~.NotificationType.NTFY`.
:param int, optional ntfyPriority: Notification option for ``type`` :attr:`~.NotificationType.NTFY`.
:param str ntfyIcon: Notification option for ``type`` :attr:`~.NotificationType.NTFY`.
:param str, optional ntfyserverurl: Notification option for ``type`` :attr:`~.NotificationType.NTFY`.
:param str ntfyIcon: Notification option for ``type`` :attr:`~.NotificationType.NTFY`.
:param str octopushVersion: Notification option for ``type`` :attr:`~.NotificationType.OCTOPUSH`.
:param str, optional octopushAPIKey: Notification option for ``type`` :attr:`~.NotificationType.OCTOPUSH`.
:param str, optional octopushLogin: Notification option for ``type`` :attr:`~.NotificationType.OCTOPUSH`.
@ -196,6 +199,7 @@ def notification_docstring(mode) -> str:
:param str pushoverpriority: Notification option for ``type`` :attr:`~.NotificationType.PUSHOVER`.
:param str pushovertitle: Notification option for ``type`` :attr:`~.NotificationType.PUSHOVER`.
:param str pushoverdevice: Notification option for ``type`` :attr:`~.NotificationType.PUSHOVER`.
:param int pushoverttl: Notification option for ``type`` :attr:`~.NotificationType.PUSHOVER`. Message TTL (Seconds).
:param str, optional pushyAPIKey: Notification option for ``type`` :attr:`~.NotificationType.PUSHY`.
:param str, optional pushyToken: Notification option for ``type`` :attr:`~.NotificationType.PUSHY`.
:param str rocketchannel: Notification option for ``type`` :attr:`~.NotificationType.ROCKET_CHAT`.

3
uptime_kuma_api/monitor_type.py
View file

@ -54,3 +54,6 @@ class MonitorType(str, Enum):
REDIS = "redis"
"""Redis"""
GROUP = "group"
"""Group"""

8
uptime_kuma_api/notification_providers.py
View file

@ -259,12 +259,14 @@ notification_provider_options = {
mattermosticonurl=dict(type="str", required=False),
),
NotificationType.NTFY: dict(
ntfyAuthenticationMethod=dict(type="str", required=False),
ntfyusername=dict(type="str", required=False),
ntfypassword=dict(type="str", required=False),
ntfyaccesstoken=dict(type="str", required=False),
ntfytopic=dict(type="str", required=True),
ntfyPriority=dict(type="int", required=True),
ntfyIcon=dict(type="str", required=False),
ntfyserverurl=dict(type="str", required=True),
ntfyIcon=dict(type="str", required=False),
),
NotificationType.OCTOPUSH: dict(
octopushVersion=dict(type="str", required=False),
@ -317,6 +319,7 @@ notification_provider_options = {
pushoverpriority=dict(type="str", required=False),
pushovertitle=dict(type="str", required=False),
pushoverdevice=dict(type="str", required=False),
pushoverttl=dict(type="int", required=False),
),
NotificationType.PUSHY: dict(
pushyAPIKey=dict(type="str", required=True),
@ -436,6 +439,9 @@ notification_provider_conditions = dict(
min=1,
max=5,
),
pushoverttl=dict(
min=0,
),
smseaglePriority=dict(
min=0,
max=9,