diff --git a/setup.py b/setup.py index 2401bfb..036de6b 100644 --- a/setup.py +++ b/setup.py @@ -15,10 +15,15 @@ here = os.path.abspath(os.path.dirname(__file__)) with open(os.path.join(here, "uptime_kuma_api", "__version__.py"), "r", "utf-8") as f: exec(f.read(), info) +with open("README.md", "r", "utf-8") as f: + readme = f.read() + setup( name=info["__title__"], version=info["__version__"], description="A python wrapper for the Uptime Kuma WebSocket API", + long_description=readme, + long_description_content_type="text/markdown", url="https://github.com/lucasheld/uptime-kuma-api", author=info["__author__"], author_email="lucasheld@hotmail.de", diff --git a/tests/requirements.txt b/tests/requirements.txt new file mode 100644 index 0000000..15e97f7 --- /dev/null +++ b/tests/requirements.txt @@ -0,0 +1 @@ +pyotp==2.6.0 diff --git a/tests/test_2fa.py b/tests/test_2fa.py new file mode 100644 index 0000000..d9d64e0 --- /dev/null +++ b/tests/test_2fa.py @@ -0,0 +1,56 @@ +import unittest +from urllib import parse + +import pyotp + +from uptime_kuma_test_case import UptimeKumaTestCase + + +def parse_secret(uri): + query = parse.urlsplit(uri).query + params = dict(parse.parse_qsl(query)) + return params["secret"] + + +def generate_token(secret): + totp = pyotp.TOTP(secret) + return totp.now() + + +class Test2FA(UptimeKumaTestCase): + def test_2fa(self): + # check 2fa is disabled + r = self.api.twofa_status() + self.assertEqual(r["status"], False) + + # prepare 2fa + r = self.api.prepare_2fa(self.password) + uri = r["uri"] + self.assertTrue(uri.startswith("otpauth://totp/")) + secret = parse_secret(uri) + + # verify token + token = generate_token(secret) + r = self.api.verify_token(token, self.password) + self.assertEqual(r["valid"], True) + + # save 2fa + r = self.api.save_2fa(self.password) + self.assertEqual(r["msg"], "2FA Enabled.") + + # check 2fa is enabled + r = self.api.twofa_status() + self.assertEqual(r["status"], True) + + # relogin using the totp token + self.api.logout() + token = generate_token(secret) + self.api.login(self.username, self.password, token) + + # disable 2fa + r = self.api.disable_2fa(self.password) + self.assertEqual(r["msg"], "2FA Disabled.") + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_avg_ping.py b/tests/test_avg_ping.py new file mode 100644 index 0000000..fc9b3be --- /dev/null +++ b/tests/test_avg_ping.py @@ -0,0 +1,13 @@ +import unittest + +from uptime_kuma_test_case import UptimeKumaTestCase + + +class TestAvgPing(UptimeKumaTestCase): + def test_avg_ping(self): + self.add_monitor() + self.api.avg_ping() + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_database.py b/tests/test_database.py new file mode 100644 index 0000000..640b9ce --- /dev/null +++ b/tests/test_database.py @@ -0,0 +1,16 @@ +import unittest + +from uptime_kuma_test_case import UptimeKumaTestCase + + +class TestDatabase(UptimeKumaTestCase): + def test_get_database_size(self): + r = self.api.get_database_size() + self.assertIn("size", r) + + def test_shrink_database(self): + self.api.shrink_database() + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_monitor.py b/tests/test_monitor.py index 8cc7d91..ef053a3 100644 --- a/tests/test_monitor.py +++ b/tests/test_monitor.py @@ -1,23 +1,19 @@ import unittest -from uptime_kuma_api import UptimeKumaException +from uptime_kuma_api import UptimeKumaException, MonitorType from uptime_kuma_test_case import UptimeKumaTestCase class TestMonitor(UptimeKumaTestCase): def test_monitor(self): expected_monitor = { - "type": "http", + "type": MonitorType.HTTP, "name": "monitor 1", - "url": "http://192.168.20.135" + "url": "http://127.0.0.1" } # add monitor - r = self.api.add_monitor( - type=expected_monitor["type"], - name=expected_monitor["name"], - url=expected_monitor["url"] - ) + r = self.api.add_monitor(**expected_monitor) self.assertEqual(r["msg"], "Added Successfully.") monitor_id = r["monitorID"] @@ -34,7 +30,7 @@ class TestMonitor(UptimeKumaTestCase): # edit monitor expected_monitor["type"] = "ping" expected_monitor["name"] = "monitor 1 new" - expected_monitor["hostname"] = "127.0.0.1" + expected_monitor["hostname"] = "127.0.0.10" del expected_monitor["url"] r = self.api.edit_monitor(monitor_id, **expected_monitor) self.assertEqual(r["msg"], "Saved.") @@ -49,12 +45,108 @@ class TestMonitor(UptimeKumaTestCase): r = self.api.resume_monitor(monitor_id) self.assertEqual(r["msg"], "Resumed Successfully.") + # get monitor beats + self.api.get_monitor_beats(monitor_id, 6) + # delete monitor r = self.api.delete_monitor(monitor_id) self.assertEqual(r["msg"], "Deleted Successfully.") with self.assertRaises(UptimeKumaException): self.api.get_monitor(monitor_id) + def do_test_monitor_type(self, expected_monitor): + r = self.api.add_monitor(**expected_monitor) + self.assertEqual(r["msg"], "Added Successfully.") + monitor_id = r["monitorID"] + + monitor = self.api.get_monitor(monitor_id) + self.compare(monitor, expected_monitor) + + def test_monitor_type_http(self): + expected_monitor = { + "type": MonitorType.HTTP, + "name": "monitor 1", + "url": "http://127.0.0.1" + } + self.do_test_monitor_type(expected_monitor) + + def test_monitor_type_port(self): + expected_monitor = { + "type": MonitorType.PORT, + "name": "monitor 1", + "hostname": "127.0.0.1", + "port": 8888 + } + self.do_test_monitor_type(expected_monitor) + + def test_monitor_type_ping(self): + expected_monitor = { + "type": MonitorType.PING, + "name": "monitor 1", + "hostname": "127.0.0.1", + } + self.do_test_monitor_type(expected_monitor) + + def test_monitor_type_keyword(self): + expected_monitor = { + "type": MonitorType.KEYWORD, + "name": "monitor 1", + "url": "http://127.0.0.1", + "keyword": "healthy" + } + self.do_test_monitor_type(expected_monitor) + + def test_monitor_type_dns(self): + expected_monitor = { + "type": MonitorType.DNS, + "name": "monitor 1", + "url": "http://127.0.0.1", + "hostname": "127.0.0.1", + "port": 8888, + "dns_resolve_server": "1.1.1.1", + } + self.do_test_monitor_type(expected_monitor) + + def test_monitor_type_push(self): + expected_monitor = { + "type": MonitorType.PUSH, + "name": "monitor 1", + "url": "http://127.0.0.1" + } + self.do_test_monitor_type(expected_monitor) + + def test_monitor_type_steam(self): + expected_monitor = { + "type": MonitorType.STEAM, + "name": "monitor 1", + "url": "http://127.0.0.1", + "hostname": "127.0.0.1", + "port": 8888, + } + self.do_test_monitor_type(expected_monitor) + + def test_monitor_type_mqtt(self): + expected_monitor = { + "type": MonitorType.MQTT, + "name": "monitor 1", + "url": "http://127.0.0.1", + "hostname": "127.0.0.1", + "port": 8888, + "mqttTopic": "test" + } + self.do_test_monitor_type(expected_monitor) + + def test_monitor_type_sqlserver(self): + expected_monitor = { + "type": MonitorType.SQLSERVER, + "name": "monitor 1", + "url": "http://127.0.0.1", + "databaseConnectionString": "Server=127.0.0.1,8888;Database=test;User Id=1;Password=secret123;Encrypt=true;" + "TrustServerCertificate=Yes;Connection Timeout=5", + "databaseQuery": "select getdate()" + } + self.do_test_monitor_type(expected_monitor) + if __name__ == '__main__': unittest.main() diff --git a/tests/test_settings.py b/tests/test_settings.py index 2777d69..39179af 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -1,4 +1,5 @@ import unittest +import json from uptime_kuma_test_case import UptimeKumaTestCase @@ -13,6 +14,32 @@ class TestSettings(UptimeKumaTestCase): settings = self.api.get_settings() self.assertEqual(settings["checkUpdate"], expected_check_update) + def test_change_password(self): + new_password = "321terces" + + # change password + r = self.api.change_password(self.password, new_password) + self.assertEqual(r["msg"], "Password has been updated successfully.") + + # check login + r = self.api.login(self.username, new_password) + self.assertIn("token", r) + + # restore password + r = self.api.change_password(new_password, self.password) + self.assertEqual(r["msg"], "Password has been updated successfully.") + + def test_upload_backup(self): + data = { + "version": "1.17.1", + "notificationList": [], + "monitorList": [], + "proxyList": [] + } + data_str = json.dumps(data) + r = self.api.upload_backup(data_str, "overwrite") + self.assertEqual(r["msg"], "Backup successfully restored.") + if __name__ == '__main__': unittest.main() diff --git a/tests/test_status_page.py b/tests/test_status_page.py index cf56c32..79b7c0c 100644 --- a/tests/test_status_page.py +++ b/tests/test_status_page.py @@ -6,12 +6,25 @@ from uptime_kuma_test_case import UptimeKumaTestCase class TestStatusPage(UptimeKumaTestCase): def test_status_page(self): + monitor_id = self.add_monitor() + slug = "slug1" expected_status_page = { "slug": slug, "title": "status page 1", "description": "description 1", - "showPoweredBy": False + "showPoweredBy": False, + "publicGroupList": [ + { + 'name': 'Services', + 'weight': 1, + 'monitorList': [ + { + "id": monitor_id + } + ] + } + ] } # slug must be unique @@ -35,7 +48,9 @@ class TestStatusPage(UptimeKumaTestCase): status_pages = self.api.get_status_pages() status_page = self.find_by_id(status_pages, slug, "slug") self.assertIsNotNone(status_page) - self.compare(status_page, expected_status_page) + # publicGroupList and incident is not available in status pages + expected_status_page_config = {i: expected_status_page[i] for i in expected_status_page if i != "publicGroupList"} + self.compare(status_page, expected_status_page_config) # edit status page expected_status_page["title"] = "status page 1 new" @@ -52,9 +67,13 @@ class TestStatusPage(UptimeKumaTestCase): } incident = self.api.post_incident(slug, **incident_expected) self.compare(incident, incident_expected) + status_page = self.api.get_status_page(slug) + self.compare(status_page["incident"], incident) # unpin incident self.api.unpin_incident(slug) + status_page = self.api.get_status_page(slug) + self.assertIsNone(status_page["incident"]) # delete status page self.api.delete_status_page(slug) diff --git a/tests/test_uptime.py b/tests/test_uptime.py new file mode 100644 index 0000000..c165eb4 --- /dev/null +++ b/tests/test_uptime.py @@ -0,0 +1,16 @@ +import unittest + +from uptime_kuma_test_case import UptimeKumaTestCase + + +class TestUptime(UptimeKumaTestCase): + def test_uptime_without_monitor(self): + self.api.uptime() + + def test_uptime_with_monitor(self): + self.add_monitor() + self.api.uptime() + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/uptime_kuma_test_case.py b/tests/uptime_kuma_test_case.py index d28b81c..967933b 100644 --- a/tests/uptime_kuma_test_case.py +++ b/tests/uptime_kuma_test_case.py @@ -1,37 +1,62 @@ +import json import unittest -from uptime_kuma_api import UptimeKumaApi - +from uptime_kuma_api import UptimeKumaApi, Event, MonitorType token = None +def compare(subset, superset): + for key, value in subset.items(): + value2 = superset.get(key) + if type(value) == list: + for i in range(len(value)): + if not value2 or not compare(value[i], value2[i]): + return False + elif type(value) == dict: + if not compare(value, value2): + return False + else: + if value != value2: + return False + return True + + class UptimeKumaTestCase(unittest.TestCase): api = None url = "http://127.0.0.1:3001" - username = "testuser" - password = "zS7zhQSc" + username = "admin" + password = "secret123" - @classmethod - def setUpClass(cls): - cls.api = UptimeKumaApi(cls.url) + def setUp(self): + self.api = UptimeKumaApi(self.url) global token if not token: - if cls.api.need_setup(): - cls.api.setup(cls.username, cls.password) - r = cls.api.login(cls.username, cls.password) + if self.api.need_setup(): + self.api.setup(self.username, self.password) + r = self.api.login(self.username, self.password) token = r["token"] - cls.api.login_by_token(token) + self.api.login_by_token(token) - @classmethod - def tearDownClass(cls): - cls.api.logout() - cls.api.disconnect() + data = { + "version": "1.17.1", + "notificationList": [], + "monitorList": [], + "proxyList": [] + } + data_str = json.dumps(data) + r = self.api.upload_backup(data_str, "overwrite") + self.assertEqual(r["msg"], "Backup successfully restored.") + + self.api._event_data[Event.MONITOR_LIST] = {} + + def tearDown(self): + self.api.disconnect() def compare(self, superset, subset): - self.assertTrue(subset.items() <= superset.items()) + self.assertTrue(compare(subset, superset)) def find_by_id(self, objects, value, key="id"): for obj in objects: @@ -39,7 +64,7 @@ class UptimeKumaTestCase(unittest.TestCase): return obj def add_monitor(self): - r = self.api.add_monitor(type="http", name="monitor 1", url="http://127.0.0.1") + r = self.api.add_monitor(type=MonitorType.HTTP, name="monitor 1", url="http://127.0.0.1") monitor_id = r["monitorID"] return monitor_id diff --git a/uptime_kuma_api/__init__.py b/uptime_kuma_api/__init__.py index edee89e..08d1ad9 100644 --- a/uptime_kuma_api/__init__.py +++ b/uptime_kuma_api/__init__.py @@ -5,4 +5,5 @@ from .notification_providers import NotificationType, notification_provider_opti from .proxy_protocol import ProxyProtocol from .incident_style import IncidentStyle from .exceptions import UptimeKumaException +from .event import Event from .api import UptimeKumaApi diff --git a/uptime_kuma_api/api.py b/uptime_kuma_api/api.py index f18cd46..e4481e4 100644 --- a/uptime_kuma_api/api.py +++ b/uptime_kuma_api/api.py @@ -1,6 +1,7 @@ import json import time +import requests import socketio from . import AuthMethod @@ -8,6 +9,7 @@ from . import MonitorType from . import NotificationType, notification_provider_options from . import ProxyProtocol from . import IncidentStyle +from . import Event from . import UptimeKumaException @@ -225,18 +227,14 @@ def _build_status_page_data( showPoweredBy: bool = True, icon: str = "/icon.svg", - monitors: list = None + publicGroupList: list = None ): if theme not in ["light", "dark"]: raise ValueError if not domainNameList: domainNameList = [] - public_group_list = [] - if monitors: - public_group_list.append({ - "name": "Services", - "monitorList": monitors - }) + if not publicGroupList: + publicGroupList = [] config = { "id": id, "slug": slug, @@ -251,7 +249,7 @@ def _build_status_page_data( "footerText": footerText, "showPoweredBy": showPoweredBy } - return slug, config, icon, public_group_list + return slug, config, icon, publicGroupList def _check_missing_arguments(required_params, kwargs): @@ -365,38 +363,45 @@ def _check_arguments_proxy(kwargs): class UptimeKumaApi(object): def __init__(self, url): + self.url = url self.sio = socketio.Client() self._event_data: dict = { - "monitorList": None, - "notificationList": None, - "proxyList": None, - "statusPageList": None, - "heartbeatList": None, - "importantHeartbeatList": None, - "avgPing": None, - "uptime": None, - "heartbeat": None, - "info": None, + Event.MONITOR_LIST: None, + Event.NOTIFICATION_LIST: None, + Event.PROXY_LIST: None, + Event.STATUS_PAGE_LIST: None, + Event.HEARTBEAT_LIST: None, + Event.IMPORTANT_HEARTBEAT_LIST: None, + Event.AVG_PING: None, + Event.UPTIME: None, + Event.HEARTBEAT: None, + Event.INFO: None, + Event.CERT_INFO: None } - self.sio.on("connect", self._event_connect) - self.sio.on("disconnect", self._event_disconnect) - self.sio.on("monitorList", self._event_monitor_list) - self.sio.on("notificationList", self._event_notification_list) - self.sio.on("proxyList", self._event_proxy_list) - self.sio.on("statusPageList", self._event_status_page_list) - self.sio.on("heartbeatList", self._event_heartbeat_list) - self.sio.on("importantHeartbeatList", self._event_important_heartbeat_list) - self.sio.on("avgPing", self._event_avg_ping) - self.sio.on("uptime", self._event_uptime) - self.sio.on("heartbeat", self._event_heartbeat) - self.sio.on("info", self._event_info) + self.sio.on(Event.CONNECT, self._event_connect) + self.sio.on(Event.DISCONNECT, self._event_disconnect) + self.sio.on(Event.MONITOR_LIST, self._event_monitor_list) + self.sio.on(Event.NOTIFICATION_LIST, self._event_notification_list) + self.sio.on(Event.PROXY_LIST, self._event_proxy_list) + self.sio.on(Event.STATUS_PAGE_LIST, self._event_status_page_list) + self.sio.on(Event.HEARTBEAT_LIST, self._event_heartbeat_list) + self.sio.on(Event.IMPORTANT_HEARTBEAT_LIST, self._event_important_heartbeat_list) + self.sio.on(Event.AVG_PING, self._event_avg_ping) + self.sio.on(Event.UPTIME, self._event_uptime) + self.sio.on(Event.HEARTBEAT, self._event_heartbeat) + self.sio.on(Event.INFO, self._event_info) + self.sio.on(Event.CERT_INFO, self._event_cert_info) - self.connect(url) + self.connect() def _get_event_data(self, event): + monitor_events = [Event.AVG_PING, Event.UPTIME, Event.HEARTBEAT_LIST, Event.IMPORTANT_HEARTBEAT_LIST, Event.CERT_INFO, Event.HEARTBEAT] while self._event_data[event] is None: + # do not wait for events that are not sent + if self._event_data[Event.MONITOR_LIST] == {} and event in monitor_events: + return [] time.sleep(0.01) time.sleep(0.01) # wait for multiple messages return self._event_data[event] @@ -418,65 +423,76 @@ class UptimeKumaApi(object): pass def _event_monitor_list(self, data): - self._event_data["monitorList"] = data + self._event_data[Event.MONITOR_LIST] = data def _event_notification_list(self, data): - self._event_data["notificationList"] = data + self._event_data[Event.NOTIFICATION_LIST] = data def _event_proxy_list(self, data): - self._event_data["proxyList"] = data + self._event_data[Event.PROXY_LIST] = data def _event_status_page_list(self, data): - self._event_data["statusPageList"] = data + self._event_data[Event.STATUS_PAGE_LIST] = data def _event_heartbeat_list(self, id_, data, bool_): - if self._event_data["heartbeatList"] is None: - self._event_data["heartbeatList"] = [] - self._event_data["heartbeatList"].append({ + if self._event_data[Event.HEARTBEAT_LIST] is None: + self._event_data[Event.HEARTBEAT_LIST] = [] + self._event_data[Event.HEARTBEAT_LIST].append({ "id": id_, "data": data, "bool": bool_, }) def _event_important_heartbeat_list(self, id_, data, bool_): - if self._event_data["importantHeartbeatList"] is None: - self._event_data["importantHeartbeatList"] = [] - self._event_data["importantHeartbeatList"].append({ + if self._event_data[Event.IMPORTANT_HEARTBEAT_LIST] is None: + self._event_data[Event.IMPORTANT_HEARTBEAT_LIST] = [] + self._event_data[Event.IMPORTANT_HEARTBEAT_LIST].append({ "id": id_, "data": data, "bool": bool_, }) def _event_avg_ping(self, id_, data): - if self._event_data["avgPing"] is None: - self._event_data["avgPing"] = [] - self._event_data["avgPing"].append({ + if self._event_data[Event.AVG_PING] is None: + self._event_data[Event.AVG_PING] = [] + self._event_data[Event.AVG_PING].append({ "id": id_, "data": data, }) def _event_uptime(self, id_, hours_24, days_30): - if self._event_data["uptime"] is None: - self._event_data["uptime"] = [] - self._event_data["uptime"].append({ + if self._event_data[Event.UPTIME] is None: + self._event_data[Event.UPTIME] = [] + self._event_data[Event.UPTIME].append({ "id": id_, "hours_24": hours_24, "days_30": days_30, }) def _event_heartbeat(self, data): - if self._event_data["heartbeat"] is None: - self._event_data["heartbeat"] = [] - self._event_data["heartbeat"].append(data) + if self._event_data[Event.HEARTBEAT] is None: + self._event_data[Event.HEARTBEAT] = [] + self._event_data[Event.HEARTBEAT].append(data) def _event_info(self, data): - self._event_data["info"] = data + self._event_data[Event.INFO] = data + + def _event_cert_info(self, id_, data): + if self._event_data[Event.CERT_INFO] is None: + self._event_data[Event.CERT_INFO] = [] + self._event_data[Event.CERT_INFO].append({ + "id": id_, + "data": data, + }) # connection - def connect(self, url: str): - url = url.rstrip("/") - self.sio.connect(f'{url}/socket.io/') + def connect(self): + url = self.url.rstrip("/") + try: + self.sio.connect(f'{url}/socket.io/') + except: + print("") def disconnect(self): self.sio.disconnect() @@ -484,7 +500,7 @@ class UptimeKumaApi(object): # monitors def get_monitors(self): - r = list(self._get_event_data("monitorList").values()) + r = list(self._get_event_data(Event.MONITOR_LIST).values()) int_to_bool(r, ["active"]) return r @@ -535,7 +551,7 @@ class UptimeKumaApi(object): # notifications def get_notifications(self): - notifications = self._get_event_data("notificationList") + notifications = self._get_event_data(Event.NOTIFICATION_LIST) r = [] for notification_raw in notifications: notification = notification_raw.copy() @@ -589,7 +605,7 @@ class UptimeKumaApi(object): # proxy def get_proxies(self): - r = self._get_event_data("proxyList") + r = self._get_event_data(Event.PROXY_LIST) int_to_bool(r, ["auth", "active", "default", "applyExisting"]) return r @@ -618,15 +634,20 @@ class UptimeKumaApi(object): # status page def get_status_pages(self): - r = list(self._get_event_data("statusPageList").values()) + r = list(self._get_event_data(Event.STATUS_PAGE_LIST).values()) return r def get_status_page(self, slug: str): - r = self._call('getStatusPage', slug) - config = r["config"] - del r["config"] - r.update(config) - return r + r1 = self._call('getStatusPage', slug) + r2 = requests.get(f"{self.url}/api/status-page/{slug}").json() + + config = r1["config"] + config.update(r2["config"]) + return { + **config, + "incident": r2["incident"], + "publicGroupList": r2["publicGroupList"] + } def add_status_page(self, slug: str, title: str): return self._call('addStatusPage', (title, slug)) @@ -636,6 +657,7 @@ class UptimeKumaApi(object): def save_status_page(self, slug: str, **kwargs): status_page = self.get_status_page(slug) + status_page.pop("incident") status_page.update(kwargs) data = _build_status_page_data(**status_page) return self._call('saveStatusPage', data) @@ -664,36 +686,41 @@ class UptimeKumaApi(object): # heartbeat def get_heartbeats(self): - r = self._get_event_data("heartbeatList") + r = self._get_event_data(Event.HEARTBEAT_LIST) for i in r: int_to_bool(i["data"], ["important", "status"]) return r def get_important_heartbeats(self): - r = self._get_event_data("importantHeartbeatList") + r = self._get_event_data(Event.IMPORTANT_HEARTBEAT_LIST) for i in r: int_to_bool(i["data"], ["important", "status"]) return r def get_heartbeat(self): - r = self._get_event_data("heartbeat") + r = self._get_event_data(Event.HEARTBEAT) int_to_bool(r, ["important", "status"]) return r # avg ping def avg_ping(self): - return self._get_event_data("avgPing") + return self._get_event_data(Event.AVG_PING) + + # cert info + + def cert_info(self): + return self._get_event_data(Event.CERT_INFO) # uptime def uptime(self): - return self._get_event_data("uptime") + return self._get_event_data(Event.UPTIME) # info def info(self): - r = self._get_event_data("info") + r = self._get_event_data(Event.INFO) return r # clear @@ -788,7 +815,7 @@ class UptimeKumaApi(object): "newPassword": new_password, }) - def upload_backup(self, json_data, import_handle: str): + def upload_backup(self, json_data, import_handle: str = "skip"): if import_handle not in ["overwrite", "skip", "keep"]: raise ValueError() return self._call('uploadBackup', (json_data, import_handle)) @@ -801,6 +828,9 @@ class UptimeKumaApi(object): def prepare_2fa(self, password: str): return self._call('prepare2FA', password) + def verify_token(self, token: str, password: str): + return self._call('verifyToken', (token, password)) + def save_2fa(self, password: str): return self._call('save2FA', password) @@ -809,19 +839,16 @@ class UptimeKumaApi(object): # login - def login(self, username: str, password: str): + def login(self, username: str, password: str, token: str = ""): return self._call('login', { "username": username, "password": password, - "token": "" + "token": token }) def login_by_token(self, token: str): return self._call('loginByToken', token) - def verify_token(self, token: str, password: str): - return self._call('verifyToken', (token, password)) - def logout(self): return self._call('logout') diff --git a/uptime_kuma_api/event.py b/uptime_kuma_api/event.py new file mode 100644 index 0000000..ad954bd --- /dev/null +++ b/uptime_kuma_api/event.py @@ -0,0 +1,17 @@ +from enum import Enum + + +class Event(str, Enum): + CONNECT = "connect" + DISCONNECT = "disconnect" + MONITOR_LIST = "monitorList" + NOTIFICATION_LIST = "notificationList" + PROXY_LIST = "proxyList" + STATUS_PAGE_LIST = "statusPageList" + HEARTBEAT_LIST = "heartbeatList" + IMPORTANT_HEARTBEAT_LIST = "importantHeartbeatList" + AVG_PING = "avgPing" + UPTIME = "uptime" + HEARTBEAT = "heartbeat" + INFO = "info" + CERT_INFO = "certInfo"