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.20.0

Browse source
This commit is contained in:
lucasheld 2023-02-13 22:51:21 +01:00
parent 8427deffe0
commit 14e9f47406
9 changed files with 276 additions and 62 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
run_tests.sh
View file

@ -5,7 +5,7 @@ if [ $version ]
then
versions=("$version")
else
versions=(1.19.5 1.18.5 1.17.1)
versions=(1.20.0 1.19.6 1.18.5 1.17.1)
fi
for version in ${versions[*]}

17
scripts/build_models.py
View file

@ -4,6 +4,9 @@ from pprint import pprint
from utils import deduplicate_list
ROOT = "uptime-kuma"
def parse_json_keys(data):
keys = []
for line in data.split("\n"):
@ -29,7 +32,7 @@ def parse_json_keys(data):
def parse_heartbeat():
with open('uptime-kuma/server/model/heartbeat.js') as f:
with open(f'{ROOT}/server/model/heartbeat.js') as f:
content = f.read()
all_keys = []
match = re.search(r'toJSON\(\) {\s+return.*{([^}]+)}', content)
@ -45,7 +48,7 @@ def parse_heartbeat():
def parse_incident():
with open('uptime-kuma/server/model/incident.js') as f:
with open(f'{ROOT}/server/model/incident.js') as f:
content = f.read()
match = re.search(r'toPublicJSON\(\) {\s+return.*{([^}]+)}', content)
data = match.group(1)
@ -55,7 +58,7 @@ def parse_incident():
def parse_monitor():
# todo: toPublicJSON ???
with open('uptime-kuma/server/model/monitor.js') as f:
with open(f'{ROOT}/server/model/monitor.js') as f:
content = f.read()
matches = re.findall(r'data = {([^}]+)}', content)
all_keys = []
@ -68,7 +71,7 @@ def parse_monitor():
def parse_proxy():
with open('uptime-kuma/server/model/proxy.js') as f:
with open(f'{ROOT}/server/model/proxy.js') as f:
content = f.read()
match = re.search(r'toJSON\(\) {\s+return.*{([^}]+)}', content)
data = match.group(1)
@ -99,7 +102,7 @@ def parse_proxy():
# # input (add, edit proxy)
# def parse_proxy2():
# with open('uptime-kuma/server/proxy.js') as f:
# with open(f'{ROOT}/server/proxy.js') as f:
# content = f.read()
#
# code = parse_function(r'async save\([^)]+\) ', content)
@ -108,7 +111,7 @@ def parse_proxy():
def parse_status_page():
with open('uptime-kuma/server/model/status_page.js') as f:
with open(f'{ROOT}/server/model/status_page.js') as f:
content = f.read()
all_keys = []
match = re.search(r'toJSON\(\) {\s+return.*{([^}]+)}', content)
@ -124,7 +127,7 @@ def parse_status_page():
def parse_tag():
with open('uptime-kuma/server/model/tag.js') as f:
with open(f'{ROOT}/server/model/tag.js') as f:
content = f.read()
match = re.search(r'toJSON\(\) {\s+return.*{([^}]+)}', content)
data = match.group(1)

21
tests/test_game_list.py Normal file
View file

@ -0,0 +1,21 @@
import unittest
from packaging.version import parse as parse_version
from uptime_kuma_test_case import UptimeKumaTestCase
class TestGameList(UptimeKumaTestCase):
def setUp(self):
super(TestGameList, self).setUp()
if parse_version(self.api.version) < parse_version("1.20"):
super(TestGameList, self).tearDown()
self.skipTest("Unsupported in this Uptime Kuma version")
def test_game_list(self):
game_list = self.api.get_game_list()
self.assertTrue("keys" in game_list[0])
if __name__ == '__main__':
unittest.main()

41
tests/test_monitor.py
View file

@ -147,6 +147,10 @@ class TestMonitor(UptimeKumaTestCase):
"name": "monitor 1",
"hostname": "127.0.0.1"
}
if parse_version(self.api.version) >= parse_version("1.20"):
expected_monitor.update({
"packetSize": 56
})
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_keyword(self):
@ -215,6 +219,21 @@ class TestMonitor(UptimeKumaTestCase):
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_gamedig(self):
if parse_version(self.api.version) < parse_version("1.20"):
self.skipTest("Unsupported in this Uptime Kuma version")
game_list = self.api.get_game_list()
game = game_list[0]["keys"][0]
expected_monitor = {
"type": MonitorType.GAMEDIG,
"name": "monitor 1",
"hostname": "127.0.0.1",
"port": 8888,
"game": game
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_mqtt(self):
expected_monitor = {
"type": MonitorType.MQTT,
@ -262,6 +281,17 @@ class TestMonitor(UptimeKumaTestCase):
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_mongodb(self):
if parse_version(self.api.version) < parse_version("1.20"):
self.skipTest("Unsupported in this Uptime Kuma version")
expected_monitor = {
"type": MonitorType.MONGODB,
"name": "monitor 1",
"databaseConnectionString": "mongodb://username:password@host:port/database"
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_radius(self):
if parse_version(self.api.version) < parse_version("1.18"):
self.skipTest("Unsupported in this Uptime Kuma version")
@ -277,6 +307,17 @@ class TestMonitor(UptimeKumaTestCase):
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_redis(self):
if parse_version(self.api.version) < parse_version("1.20"):
self.skipTest("Unsupported in this Uptime Kuma version")
expected_monitor = {
"type": MonitorType.REDIS,
"name": "monitor 1",
"databaseConnectionString": "redis://user:password@host:port"
}
self.do_test_monitor_type(expected_monitor)
if __name__ == '__main__':
unittest.main()

6
tests/test_status_page.py
View file

@ -1,5 +1,7 @@
import unittest
from packaging.version import parse as parse_version
from uptime_kuma_api import UptimeKumaException, IncidentStyle
from uptime_kuma_test_case import UptimeKumaTestCase
@ -36,6 +38,10 @@ class TestStatusPage(UptimeKumaTestCase):
}
]
}
if parse_version(self.api.version) >= parse_version("1.20"):
expected_status_page.update({
"googleAnalyticsId": ""
})
# add status page
r = self.api.add_status_page(slug, expected_status_page["title"])

11
tests/test_tag.py
View file

@ -1,5 +1,7 @@
import unittest
from packaging.version import parse as parse_version
from uptime_kuma_api import UptimeKumaException
from uptime_kuma_test_case import UptimeKumaTestCase
@ -26,6 +28,15 @@ class TestTag(UptimeKumaTestCase):
self.assertIsNotNone(tag)
self.compare(tag, expected_tag)
if parse_version(self.api.version) >= parse_version("1.20"):
# edit tag
expected_tag["name"] = "tag 1 new"
expected_tag["color"] = "#000000"
r = self.api.edit_tag(tag_id, **expected_tag)
self.assertEqual(r["msg"], "Saved")
tag = self.api.get_tag(tag_id)
self.compare(tag, expected_tag)
# delete tag
r = self.api.delete_tag(tag_id)
self.assertEqual(r["msg"], "Deleted Successfully.")

221
uptime_kuma_api/api.py
View file

@ -21,7 +21,7 @@ from . import NotificationType, notification_provider_options, notification_prov
from . import ProxyProtocol
from . import UptimeKumaException
from .docstrings import append_docstring, monitor_docstring, notification_docstring, proxy_docstring, \
docker_host_docstring, maintenance_docstring
docker_host_docstring, maintenance_docstring, tag_docstring
def int_to_bool(data, keys) -> None:
@ -59,6 +59,12 @@ def _convert_monitor_input(kwargs) -> None:
kwargs["databaseConnectionString"] = "Server=<hostname>,<port>;Database=<your database>;User Id=<your user id>;Password=<your password>;Encrypt=<true/false>;TrustServerCertificate=<Yes/No>;Connection Timeout=<int>"
elif kwargs["type"] == MonitorType.POSTGRES:
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"
if kwargs["type"] == MonitorType.PUSH and not kwargs.get("pushToken"):
kwargs["pushToken"] = gen_secret(10)
@ -125,6 +131,7 @@ def _build_status_page_data(
published: bool = True,
showTags: bool = False,
domainNameList: list = None,
googleAnalyticsId: str = None,
customCSS: str = "",
footerText: str = None,
showPoweredBy: bool = True,
@ -148,6 +155,7 @@ def _build_status_page_data(
"published": published,
"showTags": showTags,
"domainNameList": domainNameList,
"googleAnalyticsId": googleAnalyticsId,
"customCSS": customCSS,
"footerText": footerText,
"showPoweredBy": showPoweredBy
@ -219,6 +227,18 @@ def _build_maintenance_data(
return data
def _build_tag_data(
name: str,
color: str
) -> dict:
data = {
"new": True,
"name": name,
"color": color
}
return data
def _check_missing_arguments(required_params, kwargs) -> None:
missing_arguments = []
for required_param in required_params:
@ -264,11 +284,14 @@ def _check_arguments_monitor(kwargs) -> None:
MonitorType.DOCKER: ["docker_container", "docker_host"],
MonitorType.PUSH: [],
MonitorType.STEAM: ["hostname", "port"],
MonitorType.GAMEDIG: ["game", "hostname", "port"],
MonitorType.MQTT: ["hostname", "port", "mqttTopic"],
MonitorType.SQLSERVER: [],
MonitorType.POSTGRES: [],
MonitorType.MYSQL: [],
MonitorType.RADIUS: ["radiusUsername", "radiusPassword", "radiusSecret", "radiusCalledStationId", "radiusCallingStationId"]
MonitorType.MONGODB: [],
MonitorType.RADIUS: ["radiusUsername", "radiusPassword", "radiusSecret", "radiusCalledStationId", "radiusCallingStationId"],
MonitorType.REDIS: []
}
type_ = kwargs["type"]
required_args = required_args_by_type[type_]
@ -339,6 +362,14 @@ def _check_arguments_maintenance(kwargs) -> None:
_check_argument_conditions(conditions, kwargs)
def _check_arguments_tag(kwargs) -> None:
required_args = [
"name",
"color"
]
_check_missing_arguments(required_args, kwargs)
class UptimeKumaApi(object):
"""This class is used to communicate with Uptime Kuma.
@ -608,6 +639,9 @@ class UptimeKumaApi(object):
# DNS, PING, STEAM, MQTT
hostname: str = None,
# PING
packetSize: int = 56,
# DNS, STEAM, MQTT, RADIUS
port: int = None,
@ -621,8 +655,10 @@ class UptimeKumaApi(object):
mqttTopic: str = None,
mqttSuccessMessage: str = None,
# SQLSERVER, POSTGRES
# SQLSERVER, POSTGRES, MYSQL, MONGODB, REDIS
databaseConnectionString: str = None,
# SQLSERVER, POSTGRES, MYSQL
databaseQuery: str = None,
# DOCKER
@ -634,7 +670,10 @@ class UptimeKumaApi(object):
radiusPassword: str = None,
radiusSecret: str = None,
radiusCalledStationId: str = None,
radiusCallingStationId: str = None
radiusCallingStationId: str = None,
# GAMEDIG
game: str = None
) -> dict:
data = {
"type": type,
@ -699,6 +738,12 @@ class UptimeKumaApi(object):
"hostname": hostname,
})
# PING
if parse_version(self.version) >= parse_version("1.20"):
data.update({
"packetSize": packetSize,
})
# PORT, DNS, STEAM, MQTT, RADIUS
if not port:
if type == MonitorType.DNS:
@ -723,10 +768,12 @@ class UptimeKumaApi(object):
"mqttSuccessMessage": mqttSuccessMessage,
})
# SQLSERVER, POSTGRES
# SQLSERVER, POSTGRES, MYSQL, MONGODB, REDIS
data.update({
"databaseConnectionString": databaseConnectionString
})
# SQLSERVER, POSTGRES, MYSQL
if type in [MonitorType.SQLSERVER, MonitorType.POSTGRES, MonitorType.MYSQL]:
data.update({
"databaseQuery": databaseQuery,
@ -749,6 +796,12 @@ class UptimeKumaApi(object):
"radiusCallingStationId": radiusCallingStationId
})
# GAMEDIG
if type == MonitorType.GAMEDIG:
data.update({
"game": game
})
return data
# monitor
@ -781,6 +834,7 @@ class UptimeKumaApi(object):
'docker_container': None,
'docker_host': None,
'expiryNotification': False,
'game': None,
'grpcBody': None,
'grpcEnableTls': False,
'grpcMetadata': None,
@ -805,6 +859,7 @@ class UptimeKumaApi(object):
'mqttUsername': None,
'name': 'monitor 1',
'notificationIDList': [1, 2],
'packetSize': 56,
'port': None,
'proxyId': None,
'pushToken': None,
@ -858,6 +913,7 @@ class UptimeKumaApi(object):
'docker_container': None,
'docker_host': None,
'expiryNotification': False,
'game': None,
'grpcBody': None,
'grpcEnableTls': False,
'grpcMetadata': None,
@ -882,6 +938,7 @@ class UptimeKumaApi(object):
'mqttUsername': None,
'name': 'monitor 1',
'notificationIDList': [1, 2],
'packetSize': 56,
'port': None,
'proxyId': None,
'pushToken': None,
@ -1002,6 +1059,46 @@ class UptimeKumaApi(object):
int_to_bool(r, ["important", "status"])
return r
def get_game_list(self) -> list:
"""
Get a list of games that are supported by the GameDig monitor type.
:return: The server response.
:rtype: list
:raises UptimeKumaException: If the server returns an error.
Example::
>>> api.get_game_list()
[
{
'extra': {},
'keys': ['7d2d'],
'options': {
'port': 26900,
'port_query_offset': 1,
'protocol': 'valve'
},
'pretty': '7 Days to Die (2013)'
},
{
'extra': {},
'keys': ['arma2'],
'options': {
'port': 2302,
'port_query_offset': 1,
'protocol': 'valve'
},
'pretty': 'ARMA 2 (2009)'
},
...
]
"""
r = self._call('getGameList')
if not r: # workaround, gamelist is not available on first call. TODO: remove when fixed
r = self._call('getGameList')
return r["gameList"]
@append_docstring(monitor_docstring("add"))
def add_monitor(self, **kwargs) -> dict:
"""
@ -1041,7 +1138,9 @@ class UptimeKumaApi(object):
Example::
>>> api.edit_monitor(1, interval=20)
>>> api.edit_monitor(1,
... interval=20
... )
{
'monitorID': 1,
'msg': 'Saved.'
@ -1247,7 +1346,7 @@ class UptimeKumaApi(object):
Example::
>>> api.edit_notification(
>>> api.edit_notification(1,
... name="notification 1 edited",
... isDefault=False,
... applyExisting=False,
@ -1472,6 +1571,7 @@ class UptimeKumaApi(object):
'domainNameList': [],
'footerText': None,
'icon': '/icon.svg',
'googleAnalyticsId': '',
'id': 1,
'published': True,
'showPoweredBy': False,
@ -1502,6 +1602,7 @@ class UptimeKumaApi(object):
'domainNameList': [],
'footerText': None,
'icon': '/icon.svg',
'googleAnalyticsId': '',
'id': 1,
'incident': {
'content': 'content 1',
@ -1609,6 +1710,7 @@ class UptimeKumaApi(object):
: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
:param str, optional googleAnalyticsId: Google Analytics ID, defaults to None
:param str, optional customCSS: Custom CSS, defaults to ""
:param str, optional footerText: Custom Footer, defaults to None
:param bool, optional showPoweredBy: Show Powered By, defaults to True
@ -1625,14 +1727,6 @@ class UptimeKumaApi(object):
... slug="slug1",
... title="status page 1",
... description="description 1",
... theme="light",
... published=True,
... showTags=False,
... domainNameList=[],
... customCSS="",
... footerText=None,
... showPoweredBy=False,
... icon="/icon.svg",
... publicGroupList=[
... {
... 'name': 'Services',
@ -2035,13 +2129,60 @@ class UptimeKumaApi(object):
return tag
raise UptimeKumaException("tag does not exist")
# not working, monitor id required?
# def edit_tag(self, id_: int, name: str, color: str):
# return self._call('editTag', {
# "id": id_,
# "name": name,
# "color": color
# })
@append_docstring(tag_docstring("add"))
def add_tag(self, **kwargs) -> dict:
"""
Add a tag.
:return: The server response.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example::
>>> api.add_tag(
... name="tag 1",
... color="#ffffff"
... )
{
'color': '#ffffff',
'id': 1,
'name': 'tag 1'
}
"""
data = _build_tag_data(**kwargs)
_check_arguments_tag(data)
return self._call('addTag', data)["tag"]
@append_docstring(tag_docstring("edit"))
def edit_tag(self, id_: int, **kwargs) -> dict:
"""
Edits an existing tag.
:param int id_: Id of the tag to edit.
:return: The server response.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example::
>>> api.edit_tag(1,
... name="tag 1 new",
... color="#000000"
... )
{
'msg': 'Saved',
'tag': {
'id': 1,
'name': 'tag 1 new',
'color': '#000000'
}
}
"""
data = self.get_tag(id_)
data.update(kwargs)
_check_arguments_tag(data)
return self._call('editTag', data)
def delete_tag(self, id_: int) -> dict:
"""
@ -2061,34 +2202,6 @@ class UptimeKumaApi(object):
"""
return self._call('deleteTag', id_)
def add_tag(self, name: str, color: str) -> dict:
"""
Add a tag.
:param str name: Tag name
:param str color: Tag color
:return: The server response.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example::
>>> api.add_tag(
... name="tag 1",
... color="#ffffff"
... )
{
'color': '#ffffff',
'id': 1,
'name': 'tag 1'
}
"""
return self._call('addTag', {
"name": name,
"color": color,
"new": True
})["tag"]
# settings
def get_settings(self) -> dict:
@ -2635,7 +2748,9 @@ class UptimeKumaApi(object):
Example::
>>> api.edit_docker_host(1, name="name 2")
>>> api.edit_docker_host(1,
... name="name 2"
... )
{
'id': 1,
'msg': 'Saved'
@ -2666,7 +2781,6 @@ class UptimeKumaApi(object):
with self.wait_for_event(Event.DOCKER_HOST_LIST):
return self._call('deleteDockerHost', id_)
def get_maintenances(self) -> list:
"""
Get all maintenances.
@ -2967,8 +3081,7 @@ class UptimeKumaApi(object):
Example::
>>> api.edit_maintenance(
... 1,
>>> api.edit_maintenance(1,
... title="test",
... description="test",
... strategy=MaintenanceStrategy.RECURRING_INTERVAL,

10
uptime_kuma_api/docstrings.py
View file

@ -40,6 +40,7 @@ def monitor_docstring(mode) -> str:
:param str, optional authWorkstation: Workstation, defaults to None
:param str, optional keyword: Keyword. Search keyword in plain HTML or JSON response. The search is case-sensitive., defaults to None
:param str, optional hostname: Hostname, defaults to None
:param int, optional packetSize: Packet Size, defaults to None
:param int, optional port: Port, ``type`` :attr:`~.MonitorType.DNS` defaults to ``53`` and ``type`` :attr:`~.MonitorType.RADIUS` defaults to ``1812``
:param str, optional dns_resolve_server: Resolver Server, defaults to "1.1.1.1"
:param str, optional dns_resolve_type: Resource Record Type, defaults to "A"
@ -56,6 +57,7 @@ def monitor_docstring(mode) -> str:
:param str, optional radiusSecret: Radius Secret. Shared Secret between client and server., defaults to None
:param str, optional radiusCalledStationId: Called Station Id. Identifier of the called device., defaults to None
:param str, optional radiusCallingStationId: Calling Station Id. Identifier of the calling device., defaults to None
:param str, optional game: Game, defaults to None
"""
@ -271,6 +273,7 @@ def docker_host_docstring(mode) -> str:
:param str, optional dockerDaemon: Docker Daemon, defaults to None
"""
def maintenance_docstring(mode) -> str:
return f"""
:param str{", optional" if mode == "edit" else ""} title: Title
@ -283,3 +286,10 @@ def maintenance_docstring(mode) -> str:
:param list, optional daysOfMonth: List that contains the days of the month on which the maintenance is enabled (Day 1 = ``1``, Day 2 = ``2``, ..., Day 31 = ``31``) and the last day of the month (Last Day of Month = ``"lastDay1"``, 2nd Last Day of Month = ``"lastDay2"``, 3rd Last Day of Month = ``"lastDay3"``, 4th Last Day of Month = ``"lastDay4"``). Required for ``strategy`` :attr:`~.MaintenanceStrategy.RECURRING_DAY_OF_MONTH`., defaults to ``[]``.
:param list, optional timeRange: Maintenance Time Window of a Day, defaults to ``[{{"hours": 2, "minutes": 0}}, {{"hours": 3, "minutes": 0}}]``.
"""
def tag_docstring(mode) -> str:
return f"""
:param str{", optional" if mode == "edit" else ""} name: Tag name
:param str{", optional" if mode == "edit" else ""} color: Tag color
"""

9
uptime_kuma_api/monitor_type.py
View file

@ -31,6 +31,9 @@ class MonitorType(str, Enum):
STEAM = "steam"
"""Steam Game Server"""
GAMEDIG = "gamedig"
"""GameDig"""
MQTT = "mqtt"
"""MQTT"""
@ -43,5 +46,11 @@ class MonitorType(str, Enum):
MYSQL = "mysql"
"""MySQL/MariaDB"""
MONGODB = "mongodb"
"""MongoDB"""
RADIUS = "radius"
"""Radius"""
REDIS = "redis"
"""Redis"""