feat: add support for uptime kuma 1.21.0

This commit is contained in:
lucasheld 2023-03-20 15:14:39 +01:00
parent 550be17817
commit 42c040f451
8 changed files with 329 additions and 12 deletions

View file

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

62
tests/test_api_key.py Normal file
View file

@ -0,0 +1,62 @@
import unittest
from packaging.version import parse as parse_version
from uptime_kuma_api import DockerType, UptimeKumaException
from uptime_kuma_test_case import UptimeKumaTestCase
class TestApiKey(UptimeKumaTestCase):
def setUp(self):
super(TestApiKey, self).setUp()
if parse_version(self.api.version) < parse_version("1.21"):
super(TestApiKey, self).tearDown()
self.skipTest("Unsupported in this Uptime Kuma version")
def test_api_key(self):
# get empty list to make sure that future accesses will also work
self.api.get_api_keys()
expected = {
"name": "name 1",
"expires": "2023-03-30 12:20:00",
"active": True
}
# add api key
r = self.api.add_api_key(**expected)
self.assertEqual(r["msg"], "Added Successfully.")
api_key_id = r["keyID"]
# get api key
api_key = self.api.get_api_key(api_key_id)
self.compare(api_key, expected)
# get api keys
api_keys = self.api.get_api_keys()
api_key = self.find_by_id(api_keys, api_key_id)
self.assertIsNotNone(api_key)
self.compare(api_key, expected)
# disable api key
r = self.api.disable_api_key(api_key_id)
self.assertEqual(r["msg"], "Disabled Successfully.")
api_key = self.api.get_api_key(api_key_id)
expected["active"] = False
self.compare(api_key, expected)
# enable api key
r = self.api.enable_api_key(api_key_id)
self.assertEqual(r["msg"], "Enabled Successfully")
api_key = self.api.get_api_key(api_key_id)
expected["active"] = True
self.compare(api_key, expected)
# delete api key
r = self.api.delete_api_key(api_key_id)
self.assertEqual(r["msg"], "Deleted Successfully.")
with self.assertRaises(UptimeKumaException):
self.api.get_api_key(api_key_id)
if __name__ == '__main__':
unittest.main()

View file

@ -132,6 +132,18 @@ class TestMonitor(UptimeKumaTestCase):
}
self.do_test_monitor_type(expected_monitor)
if parse_version(self.api.version) >= parse_version("1.21"):
expected_monitor = {
"type": MonitorType.HTTP,
"name": "monitor 1",
"url": "http://127.0.0.1",
"authMethod": AuthMethod.MTLS,
"tlsCert": "cert",
"tlsKey": "key",
"tlsCa": "ca",
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_port(self):
expected_monitor = {
"type": MonitorType.PORT,

View file

@ -424,7 +424,8 @@ class UptimeKumaApi(object):
Event.CERT_INFO: None,
Event.DOCKER_HOST_LIST: None,
Event.AUTO_LOGIN: None,
Event.MAINTENANCE_LIST: None
Event.MAINTENANCE_LIST: None,
Event.API_KEY_LIST: None
}
self.sio.on(Event.CONNECT, self._event_connect)
@ -444,6 +445,7 @@ class UptimeKumaApi(object):
self.sio.on(Event.AUTO_LOGIN, self._event_auto_login)
self.sio.on(Event.INIT_SERVER_TIMEZONE, self._event_init_server_timezone)
self.sio.on(Event.MAINTENANCE_LIST, self._event_maintenance_list)
self.sio.on(Event.API_KEY_LIST, self._event_api_key_list)
self.connect()
@ -566,6 +568,9 @@ class UptimeKumaApi(object):
def _event_maintenance_list(self, data) -> None:
self._event_data[Event.MAINTENANCE_LIST] = data
def _event_api_key_list(self, data) -> None:
self._event_data[Event.API_KEY_LIST] = data
# connection
def connect(self) -> None:
@ -601,12 +606,14 @@ class UptimeKumaApi(object):
self,
type: MonitorType,
name: str,
description: str = None,
interval: int = 60,
retryInterval: int = 60,
resendInterval: int = 0,
maxretries: int = 0,
upsideDown: bool = False,
notificationIDList: list = None,
httpBodyEncoding: str = "json",
# HTTP, KEYWORD
url: str = None,
@ -619,6 +626,9 @@ class UptimeKumaApi(object):
body: str = None,
headers: str = None,
authMethod: AuthMethod = AuthMethod.NONE,
tlsCert: str = None,
tlsKey: str = None,
tlsCa: str = None,
basic_auth_user: str = None,
basic_auth_pass: str = None,
authDomain: str = None,
@ -690,6 +700,12 @@ class UptimeKumaApi(object):
"resendInterval": resendInterval
})
if parse_version(self.version) >= parse_version("1.21"):
data.update({
"description": description,
"httpBodyEncoding": httpBodyEncoding
})
if type in [MonitorType.KEYWORD, MonitorType.GRPC_KEYWORD]:
data.update({
"keyword": keyword,
@ -721,6 +737,13 @@ class UptimeKumaApi(object):
"authWorkstation": authWorkstation,
})
if authMethod == AuthMethod.MTLS:
data.update({
"tlsCert": tlsCert,
"tlsKey": tlsKey,
"tlsCa": tlsCa,
})
# GRPC_KEYWORD
if type == MonitorType.GRPC_KEYWORD:
data.update({
@ -878,6 +901,9 @@ class UptimeKumaApi(object):
}
]
"""
# TODO: replace with getMonitorList?
r = list(self._get_event_data(Event.MONITOR_LIST).values())
for monitor in r:
_convert_monitor_return(monitor)
@ -1095,7 +1121,10 @@ class UptimeKumaApi(object):
]
"""
r = self._call('getGameList')
if not r: # workaround, gamelist is not available on first call. TODO: remove when fixed
# Workaround, gamelist is not available on first call.
# Fixed in https://github.com/louislam/uptime-kuma/commit/7b8ed01f272fc4c6b69ff6299185e936a5e63735
# Exists in 1.20.0 - 1.21.0
if not r:
r = self._call('getGameList')
return r["gameList"]
@ -2781,12 +2810,14 @@ class UptimeKumaApi(object):
with self.wait_for_event(Event.DOCKER_HOST_LIST):
return self._call('deleteDockerHost', id_)
# maintenance
def get_maintenances(self) -> list:
"""
Get all maintenances.
:return: All maintenances.
:rtype: dict
:rtype: list
:raises UptimeKumaException: If the server returns an error.
Example::
@ -3176,7 +3207,7 @@ class UptimeKumaApi(object):
:param int id_: Id of the maintenance to get the monitors from.
:return: All monitors of the maintenance.
:rtype: dict
:rtype: list
:raises UptimeKumaException: If the server returns an error.
Example::
@ -3234,7 +3265,7 @@ class UptimeKumaApi(object):
:param int id_: Id of the maintenance to get the status pages from.
:return: All status pages of the maintenance.
:rtype: dict
:rtype: list
:raises UptimeKumaException: If the server returns an error.
Example::
@ -3281,3 +3312,172 @@ class UptimeKumaApi(object):
}
"""
return self._call('addMaintenanceStatusPage', (id_, status_pages))
# api key
def get_api_keys(self) -> list:
"""
Get all api keys.
:return: All api keys.
:rtype: list
:raises UptimeKumaException: If the server returns an error.
Example::
>>> api.get_api_key_list()
[
{
"id": 1,
"name": "test",
"userID": 1,
"createdDate": "2023-03-20 11:15:05",
"active": False,
"expires": null,
"status": "inactive"
},
{
"id": 2,
"name": "test2",
"userID": 1,
"createdDate": "2023-03-20 11:20:29",
"active": True,
"expires": "2023-03-30 12:20:00",
"status": "active"
}
]
"""
# TODO: replace with getAPIKeyList?
r = self._get_event_data(Event.API_KEY_LIST)
int_to_bool(r, ["active"])
return r
def get_api_key(self, id_: int) -> dict:
"""
Get an api key.
:param int id_: Id of the api key to get.
:return: The api key.
:rtype: dict
:raises UptimeKumaException: If the api key does not exist.
Example::
>>> api.get_api_key(1)
{
"id": 1,
"name": "test",
"userID": 1,
"createdDate": "2023-03-20 11:15:05",
"active": False,
"expires": null,
"status": "inactive"
}
"""
api_keys = self.get_api_keys()
for api_key in api_keys:
if api_key["id"] == id_:
return api_key
raise UptimeKumaException("notification does not exist")
def add_api_key(self, name: str, expires: str, active: bool) -> dict:
"""
Adds a new api key.
:param str name: Name of the api key.
:param str expires: Expiration date of the api key. Set to ``None`` to disable expiration.
:param bool active: True to activate api key.
:return: The server response.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example::
>>> api.add_api_key(
... name="test",
... expires="2023-03-30 12:20:00",
... active=True
... )
{
"msg": "Added Successfully.",
"key": "uk1_9XPRjV7ilGj9CvWRKYiBPq9GLtQs74UzTxKfCxWY",
"keyID": 1
}
>>> api.add_api_key(
... name="test2",
... expires=None,
... active=True
... )
{
"msg": "Added Successfully.",
"key": "uk2_jsB9H1Zmt9eEjycNFMTKgse1B0Vfvb944H4_aRqW",
"keyID": 2
}
"""
data = {
"name": name,
"expires": expires,
"active": 1 if active else 0
}
with self.wait_for_event(Event.API_KEY_LIST):
return self._call('addAPIKey', data)
def enable_api_key(self, id_: int) -> dict:
"""
Enable an api key.
:param int id_: Id of the api key to enable.
:return: The server response.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example::
>>> api.enable_api_key(1)
{
"msg": "Enabled Successfully"
}
"""
with self.wait_for_event(Event.API_KEY_LIST):
return self._call('enableAPIKey', id_)
def disable_api_key(self, id_: int) -> dict:
"""
Disable an api key.
:param int id_: Id of the api key to disable.
:return: The server response.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example::
>>> api.disable_api_key(1)
{
"msg": "Disabled Successfully."
}
"""
with self.wait_for_event(Event.API_KEY_LIST):
return self._call('disableAPIKey', id_)
def delete_api_key(self, id_: int) -> dict:
"""
Enable an api key.
:param int id_: Id of the api key to delete.
:return: The server response.
:rtype: dict
:raises UptimeKumaException: If the server returns an error.
Example::
>>> api.delete_api_key(1)
{
"msg": "Deleted Successfully."
}
"""
with self.wait_for_event(Event.API_KEY_LIST):
return self._call('deleteAPIKey', id_)

View file

@ -12,3 +12,6 @@ class AuthMethod(str, Enum):
NTLM = "ntlm"
"""NTLM Authentication."""
MTLS = "mtls"
"""mTLS Authentication."""

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 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
:param int, optional resendInterval: Resend every X times, defaults to 0
@ -31,13 +32,17 @@ def monitor_docstring(mode) -> str:
:param list, optional accepted_statuscodes: Accepted Status Codes. Select status codes which are considered as a successful response., defaults to None
:param int, optional proxyId: Proxy, defaults to None
:param str, optional method: Method, defaults to "GET"
:param str, optional httpBodyEncoding: Body Encoding, defaults to "json". Allowed values: "json", "xml".
:param str, optional body: Body, defaults to None
:param str, optional headers: Headers, defaults to None
:param AuthMethod, optional authMethod: Method, defaults to :attr:`~.AuthMethod.NONE`
:param str, optional basic_auth_user: Username, defaults to None
:param str, optional basic_auth_pass: Password, defaults to None
:param str, optional authDomain: Domain, defaults to None
:param str, optional authWorkstation: Workstation, defaults to None
:param str, optional tlsCert: Cert for ``authMethod`` :attr:`~.AuthMethod.MTLS`, defaults to None.
:param str, optional tlsKey: Key for ``authMethod`` :attr:`~.AuthMethod.MTLS`, defaults to None.
:param str, optional tlsCa: Ca for ``authMethod`` :attr:`~.AuthMethod.MTLS`, defaults to None.
:param str, optional basic_auth_user: Username for ``authMethod`` :attr:`~.AuthMethod.HTTP_BASIC` and :attr:`~.AuthMethod.NTLM`, defaults to None
:param str, optional basic_auth_pass: Password for ``authMethod`` :attr:`~.AuthMethod.HTTP_BASIC` and :attr:`~.AuthMethod.NTLM`, defaults to None
:param str, optional authDomain: Domain for ``authMethod`` :attr:`~.AuthMethod.NTLM`, defaults to None
:param str, optional authWorkstation: Workstation for ``authMethod`` :attr:`~.AuthMethod.NTLM`, 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
@ -103,6 +108,8 @@ def notification_docstring(mode) -> str:
:param int, optional gotifyPriority: Notification option for ``type`` :attr:`~.NotificationType.GOTIFY`
:param str, optional lineChannelAccessToken: Notification option for ``type`` :attr:`~.NotificationType.LINE`
:param str, optional lineUserID: Notification option for ``type`` :attr:`~.NotificationType.LINE`
:param str, optional lunaseaTarget: Notification option for ``type`` :attr:`~.NotificationType.LUNASEA`. Allowed values: "device", "user".
:param str, optional lunaseaUserID: Notification option for ``type`` :attr:`~.NotificationType.LUNASEA`
:param str, optional lunaseaDevice: Notification option for ``type`` :attr:`~.NotificationType.LUNASEA`
:param str, optional internalRoomId: Notification option for ``type`` :attr:`~.NotificationType.MATRIX`
:param str, optional accessToken: Notification option for ``type`` :attr:`~.NotificationType.MATRIX`
@ -120,6 +127,22 @@ def notification_docstring(mode) -> str:
:param str, optional pagerdutyIntegrationUrl: Notification option for ``type`` :attr:`~.NotificationType.PAGERDUTY`
:param str, optional pagerdutyPriority: Notification option for ``type`` :attr:`~.NotificationType.PAGERDUTY`
:param str, optional pagerdutyIntegrationKey: Notification option for ``type`` :attr:`~.NotificationType.PAGERDUTY`
:param str, optional pagertreeAutoResolve: Notification option for ``type`` :attr:`~.NotificationType.PAGERTREE`
Available values are:
- ``0``: Do Nothing
- ``resolve``: Auto Resolve
:param str, optional pagertreeIntegrationUrl: Notification option for ``type`` :attr:`~.NotificationType.PAGERTREE`
:param str, optional pagertreeUrgency: Notification option for ``type`` :attr:`~.NotificationType.PAGERTREE`.
Available values are:
- ``silent``: Silent
- ``low``: Low
- ``medium``: Medium
- ``high``: High
- ``critical``: Critical
:param str, optional promosmsLogin: Notification option for ``type`` :attr:`~.NotificationType.PROMOSMS`
:param str, optional promosmsPassword: Notification option for ``type`` :attr:`~.NotificationType.PROMOSMS`
:param str, optional promosmsPhoneNumber: Notification option for ``type`` :attr:`~.NotificationType.PROMOSMS`. Phone number (for Polish recipient You can skip area codes).
@ -179,8 +202,11 @@ def notification_docstring(mode) -> str:
:param str, optional smtpTo: Notification option for ``type`` :attr:`~.NotificationType.SMTP`
:param str, optional stackfieldwebhookURL: Notification option for ``type`` :attr:`~.NotificationType.STACKFIELD`
:param str, optional pushAPIKey: Notification option for ``type`` :attr:`~.NotificationType.PUSHBYTECHULUS`
:param str, optional telegramBotToken: Notification option for ``type`` :attr:`~.NotificationType.TELEGRAM`
:param str, optional telegramChatID: Notification option for ``type`` :attr:`~.NotificationType.TELEGRAM`
:param bool, optional telegramSendSilently: Notification option for ``type`` :attr:`~.NotificationType.TELEGRAM`
:param bool, optional telegramProtectContent: Notification option for ``type`` :attr:`~.NotificationType.TELEGRAM`
:param str, optional telegramMessageThreadID: Notification option for ``type`` :attr:`~.NotificationType.TELEGRAM`
:param str, optional telegramBotToken: Notification option for ``type`` :attr:`~.NotificationType.TELEGRAM`
:param str, optional webhookContentType: Notification option for ``type`` :attr:`~.NotificationType.WEBHOOK`
:param str, optional webhookAdditionalHeaders: Notification option for ``type`` :attr:`~.NotificationType.WEBHOOK`
:param str, optional webhookURL: Notification option for ``type`` :attr:`~.NotificationType.WEBHOOK`

View file

@ -19,3 +19,4 @@ class Event(str, Enum):
AUTO_LOGIN = "autoLogin"
INIT_SERVER_TIMEZONE = "initServerTimezone"
MAINTENANCE_LIST = "maintenanceList"
API_KEY_LIST = "apiKeyList"

View file

@ -52,6 +52,9 @@ class NotificationType(str, Enum):
PAGERDUTY = "PagerDuty"
"""PagerDuty"""
PAGERTREE = "PagerTree"
"""PagerTree"""
PROMOSMS = "promosms"
"""PromoSMS"""
@ -207,6 +210,8 @@ notification_provider_options = {
lineUserID=dict(type="str"),
),
NotificationType.LUNASEA: dict(
lunaseaTarget=dict(type="str"),
lunaseaUserID=dict(type="str"),
lunaseaDevice=dict(type="str"),
),
NotificationType.MATRIX: dict(
@ -233,6 +238,11 @@ notification_provider_options = {
pagerdutyPriority=dict(type="str"),
pagerdutyIntegrationKey=dict(type="str"),
),
NotificationType.PAGERTREE: dict(
pagertreeAutoResolve=dict(type="str"),
pagertreeIntegrationUrl=dict(type="str"),
pagertreeUrgency=dict(type="str"),
),
NotificationType.PROMOSMS: dict(
promosmsLogin=dict(type="str"),
promosmsPassword=dict(type="str"),
@ -310,8 +320,11 @@ notification_provider_options = {
pushAPIKey=dict(type="str"),
),
NotificationType.TELEGRAM: dict(
telegramBotToken=dict(type="str"),
telegramChatID=dict(type="str"),
telegramSendSilently=dict(type="bool"),
telegramProtectContent=dict(type="bool"),
telegramMessageThreadID=dict(type="str"),
telegramBotToken=dict(type="str"),
),
NotificationType.WEBHOOK: dict(
webhookContentType=dict(type="str"),