diff --git a/run_tests.sh b/run_tests.sh index 2548598..66753ae 100755 --- a/run_tests.sh +++ b/run_tests.sh @@ -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[*]} diff --git a/tests/test_api_key.py b/tests/test_api_key.py new file mode 100644 index 0000000..a0fdbc6 --- /dev/null +++ b/tests/test_api_key.py @@ -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() diff --git a/tests/test_monitor.py b/tests/test_monitor.py index 9d18139..30964ab 100644 --- a/tests/test_monitor.py +++ b/tests/test_monitor.py @@ -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, diff --git a/uptime_kuma_api/api.py b/uptime_kuma_api/api.py index b3f223c..e13ac01 100644 --- a/uptime_kuma_api/api.py +++ b/uptime_kuma_api/api.py @@ -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_) diff --git a/uptime_kuma_api/auth_method.py b/uptime_kuma_api/auth_method.py index 71b65a6..ff36b07 100644 --- a/uptime_kuma_api/auth_method.py +++ b/uptime_kuma_api/auth_method.py @@ -12,3 +12,6 @@ class AuthMethod(str, Enum): NTLM = "ntlm" """NTLM Authentication.""" + + MTLS = "mtls" + """mTLS Authentication.""" diff --git a/uptime_kuma_api/docstrings.py b/uptime_kuma_api/docstrings.py index 0b007e1..b885281 100644 --- a/uptime_kuma_api/docstrings.py +++ b/uptime_kuma_api/docstrings.py @@ -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` diff --git a/uptime_kuma_api/event.py b/uptime_kuma_api/event.py index a6469b5..240ade9 100644 --- a/uptime_kuma_api/event.py +++ b/uptime_kuma_api/event.py @@ -19,3 +19,4 @@ class Event(str, Enum): AUTO_LOGIN = "autoLogin" INIT_SERVER_TIMEZONE = "initServerTimezone" MAINTENANCE_LIST = "maintenanceList" + API_KEY_LIST = "apiKeyList" diff --git a/uptime_kuma_api/notification_providers.py b/uptime_kuma_api/notification_providers.py index 3d0a27d..5c7b3a6 100644 --- a/uptime_kuma_api/notification_providers.py +++ b/uptime_kuma_api/notification_providers.py @@ -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"),