feat: replace raw return values with enum values

BREAKING CHANGE:
Types of return values changed to enum values:
  - monitor: `type` (str -> MonitorType), `authMethod` (str -> AuthMethod)
  - notification: `type` (str -> NotificationType)
  - docker host: `dockerType` (str -> DockerType)
  - status page: `style` (str -> IncidentStyle)
  - maintenance: `strategy` (str -> MaintenanceStrategy)
  - proxy: `protocol` (str -> ProxyProtocol)
This commit is contained in:
lucasheld 2023-05-25 21:26:54 +02:00
parent 33b8ffc476
commit 84d4009d6a
8 changed files with 93 additions and 27 deletions

View file

@ -30,6 +30,7 @@ class TestDockerHost(UptimeKumaTestCase):
# get docker hosts # get docker hosts
docker_hosts = self.api.get_docker_hosts() docker_hosts = self.api.get_docker_hosts()
self.assertTrue(type(docker_hosts[0]["dockerType"]) == DockerType)
docker_host = self.find_by_id(docker_hosts, docker_host_id) docker_host = self.find_by_id(docker_hosts, docker_host_id)
self.assertIsNotNone(docker_host) self.assertIsNotNone(docker_host)
self.compare(docker_host, expected_docker_host) self.compare(docker_host, expected_docker_host)

View file

@ -1,14 +1,19 @@
import unittest import unittest
from uptime_kuma_api import MonitorStatus
from uptime_kuma_test_case import UptimeKumaTestCase from uptime_kuma_test_case import UptimeKumaTestCase
class TestHeartbeat(UptimeKumaTestCase): class TestHeartbeat(UptimeKumaTestCase):
def test_get_heartbeats(self): def test_get_heartbeats(self):
self.api.get_heartbeats() self.add_monitor()
r = self.api.get_heartbeats()
self.assertTrue(type(list(r.values())[0][0]["status"]) == MonitorStatus)
def test_get_important_heartbeats(self): def test_get_important_heartbeats(self):
self.api.get_important_heartbeats() self.add_monitor()
r = self.api.get_important_heartbeats()
self.assertTrue(type(list(r.values())[0][0]["status"]) == MonitorStatus)
if __name__ == '__main__': if __name__ == '__main__':

View file

@ -28,10 +28,12 @@ class TestMaintenance(UptimeKumaTestCase):
# get maintenance # get maintenance
maintenance = self.api.get_maintenance(maintenance_id) maintenance = self.api.get_maintenance(maintenance_id)
self.assertTrue(type(maintenance["strategy"]) == MaintenanceStrategy)
self.compare(maintenance, expected_maintenance) self.compare(maintenance, expected_maintenance)
# get maintenances # get maintenances
maintenances = self.api.get_maintenances() maintenances = self.api.get_maintenances()
self.assertTrue(type(maintenances[0]["strategy"]) == MaintenanceStrategy)
maintenance = self.find_by_id(maintenances, maintenance_id) maintenance = self.find_by_id(maintenances, maintenance_id)
self.assertIsNotNone(maintenance) self.assertIsNotNone(maintenance)
self.compare(maintenance, expected_maintenance) self.compare(maintenance, expected_maintenance)

View file

@ -1,6 +1,6 @@
import unittest import unittest
from uptime_kuma_api import UptimeKumaException, MonitorType, AuthMethod from uptime_kuma_api import UptimeKumaException, MonitorType, AuthMethod, MonitorStatus
from uptime_kuma_test_case import UptimeKumaTestCase from uptime_kuma_test_case import UptimeKumaTestCase
@ -35,7 +35,11 @@ class TestMonitor(UptimeKumaTestCase):
# get monitors # get monitors
monitors = self.api.get_monitors() monitors = self.api.get_monitors()
self.assertTrue(type(monitors[0]["type"]) == MonitorType)
self.assertTrue(type(monitors[0]["authMethod"]) == AuthMethod)
monitor = self.find_by_id(monitors, monitor_id) monitor = self.find_by_id(monitors, monitor_id)
self.assertTrue(type(monitor["type"]) == MonitorType)
self.assertTrue(type(monitor["authMethod"]) == AuthMethod)
self.assertIsNotNone(monitor) self.assertIsNotNone(monitor)
self.compare(monitor, expected_monitor) self.compare(monitor, expected_monitor)
@ -58,7 +62,8 @@ class TestMonitor(UptimeKumaTestCase):
self.assertEqual(r["msg"], "Resumed Successfully.") self.assertEqual(r["msg"], "Resumed Successfully.")
# get monitor beats # get monitor beats
self.api.get_monitor_beats(monitor_id, 6) r = self.api.get_monitor_beats(monitor_id, 6)
self.assertTrue(type(r[0]["status"]) == MonitorStatus)
# delete monitor # delete monitor
r = self.api.delete_monitor(monitor_id) r = self.api.delete_monitor(monitor_id)

View file

@ -33,7 +33,9 @@ class TestNotification(UptimeKumaTestCase):
# get notifications # get notifications
notifications = self.api.get_notifications() notifications = self.api.get_notifications()
self.assertTrue(type(notifications[0]["type"]) == NotificationType)
notification = self.find_by_id(notifications, notification_id) notification = self.find_by_id(notifications, notification_id)
self.assertTrue(type(notification["type"]) == NotificationType)
self.assertIsNotNone(notification) self.assertIsNotNone(notification)
self.compare(notification, expected_notification) self.compare(notification, expected_notification)

View file

@ -31,6 +31,7 @@ class TestProxy(UptimeKumaTestCase):
# get proxies # get proxies
proxies = self.api.get_proxies() proxies = self.api.get_proxies()
self.assertTrue(type(proxies[0]["protocol"]) == ProxyProtocol)
proxy = self.find_by_id(proxies, proxy_id) proxy = self.find_by_id(proxies, proxy_id)
self.assertIsNotNone(proxy) self.assertIsNotNone(proxy)
self.compare(proxy, expected_proxy) self.compare(proxy, expected_proxy)

View file

@ -71,9 +71,11 @@ class TestStatusPage(UptimeKumaTestCase):
"style": IncidentStyle.DANGER "style": IncidentStyle.DANGER
} }
incident = self.api.post_incident(slug, **incident_expected) incident = self.api.post_incident(slug, **incident_expected)
self.assertTrue(type(incident["style"]) == IncidentStyle)
self.compare(incident, incident_expected) self.compare(incident, incident_expected)
status_page = self.api.get_status_page(slug) status_page = self.api.get_status_page(slug)
self.compare(status_page["incident"], incident) self.compare(status_page["incident"], incident)
self.assertTrue(type(status_page["incident"]["style"]) == IncidentStyle)
# unpin incident # unpin incident
self.api.unpin_incident(slug) self.api.unpin_incident(slug)

View file

@ -49,18 +49,53 @@ def int_to_bool(data, keys) -> None:
data[key] = True if data[key] == 1 else False data[key] = True if data[key] == 1 else False
def parse_value(data, func) -> None: def parse_value(data, key, type_) -> None:
if not data:
return
if isinstance(data, list): if isinstance(data, list):
for d in data: for d in data:
parse_value(d, func) parse_value(d, key, type_)
else: else:
func(data) if key in data:
data[key] = type_(data[key])
# monitor
def parse_monitor_status(data) -> None: def parse_monitor_status(data) -> None:
def parse(x): parse_value(data, "status", MonitorStatus)
x["status"] = MonitorStatus(x["status"])
parse_value(data, parse)
def parse_monitor_type(data) -> None:
parse_value(data, "type", MonitorType)
def parse_auth_method(data) -> None:
parse_value(data, "authMethod", AuthMethod)
# notification
def parse_notification_type(data) -> None:
parse_value(data, "type", NotificationType)
# docker host
def parse_docker_type(data) -> None:
parse_value(data, "dockerType", DockerType)
# status page
def parse_incident_style(data) -> None:
parse_value(data, "style", IncidentStyle)
# maintenance
def parse_maintenance_strategy(data) -> None:
parse_value(data, "strategy", MaintenanceStrategy)
# proxy
def parse_proxy_protocol(data) -> None:
parse_value(data, "protocol", ProxyProtocol)
def gen_secret(length: int) -> str: def gen_secret(length: int) -> str:
@ -918,7 +953,7 @@ class UptimeKumaApi(object):
'accepted_statuscodes': ['200-299'], 'accepted_statuscodes': ['200-299'],
'active': True, 'active': True,
'authDomain': None, 'authDomain': None,
'authMethod': '', 'authMethod': <AuthMethod.NONE: ''>,
'authWorkstation': None, 'authWorkstation': None,
'basic_auth_pass': None, 'basic_auth_pass': None,
'basic_auth_user': None, 'basic_auth_user': None,
@ -968,7 +1003,7 @@ class UptimeKumaApi(object):
'resendInterval': 0, 'resendInterval': 0,
'retryInterval': 60, 'retryInterval': 60,
'tags': [], 'tags': [],
'type': 'http', 'type': <MonitorType.HTTP: 'http'>
'upsideDown': False, 'upsideDown': False,
'url': 'http://127.0.0.1', 'url': 'http://127.0.0.1',
'weight': 2000 'weight': 2000
@ -982,6 +1017,8 @@ class UptimeKumaApi(object):
for monitor in r: for monitor in r:
_convert_monitor_return(monitor) _convert_monitor_return(monitor)
int_to_bool(r, ["active"]) int_to_bool(r, ["active"])
parse_monitor_type(r)
parse_auth_method(r)
return r return r
def get_monitor(self, id_: int) -> dict: def get_monitor(self, id_: int) -> dict:
@ -1000,7 +1037,7 @@ class UptimeKumaApi(object):
'accepted_statuscodes': ['200-299'], 'accepted_statuscodes': ['200-299'],
'active': True, 'active': True,
'authDomain': None, 'authDomain': None,
'authMethod': '', 'authMethod': <AuthMethod.NONE: ''>,
'authWorkstation': None, 'authWorkstation': None,
'basic_auth_pass': None, 'basic_auth_pass': None,
'basic_auth_user': None, 'basic_auth_user': None,
@ -1050,7 +1087,7 @@ class UptimeKumaApi(object):
'resendInterval': 0, 'resendInterval': 0,
'retryInterval': 60, 'retryInterval': 60,
'tags': [], 'tags': [],
'type': 'http', 'type': <MonitorType.HTTP: 'http'>
'upsideDown': False, 'upsideDown': False,
'url': 'http://127.0.0.1', 'url': 'http://127.0.0.1',
'weight': 2000 'weight': 2000
@ -1059,6 +1096,8 @@ class UptimeKumaApi(object):
r = self._call('getMonitor', id_)["monitor"] r = self._call('getMonitor', id_)["monitor"]
_convert_monitor_return(r) _convert_monitor_return(r)
int_to_bool(r, ["active"]) int_to_bool(r, ["active"])
parse_monitor_type(r)
parse_auth_method(r)
return r return r
def pause_monitor(self, id_: int) -> dict: def pause_monitor(self, id_: int) -> dict:
@ -1346,7 +1385,7 @@ class UptimeKumaApi(object):
'isDefault': True, 'isDefault': True,
'name': 'notification 1', 'name': 'notification 1',
'pushAPIKey': '123456789', 'pushAPIKey': '123456789',
'type': 'PushByTechulus', 'type': <NotificationType.PUSHBYTECHULUS: 'PushByTechulus'>
'userId': 1 'userId': 1
} }
] ]
@ -1359,6 +1398,7 @@ class UptimeKumaApi(object):
del notification["config"] del notification["config"]
notification.update(config) notification.update(config)
r.append(notification) r.append(notification)
parse_notification_type(r)
return r return r
def get_notification(self, id_: int) -> dict: def get_notification(self, id_: int) -> dict:
@ -1380,7 +1420,7 @@ class UptimeKumaApi(object):
'isDefault': True, 'isDefault': True,
'name': 'notification 1', 'name': 'notification 1',
'pushAPIKey': '123456789', 'pushAPIKey': '123456789',
'type': 'PushByTechulus', 'type': <NotificationType.PUSHBYTECHULUS: 'PushByTechulus'>
'userId': 1 'userId': 1
} }
""" """
@ -1545,7 +1585,7 @@ class UptimeKumaApi(object):
'id': 1, 'id': 1,
'password': 'password', 'password': 'password',
'port': 8080, 'port': 8080,
'protocol': 'http', 'protocol': <ProxyProtocol.HTTP: 'http'>,
'userId': 1, 'userId': 1,
'username': 'username' 'username': 'username'
} }
@ -1553,6 +1593,7 @@ class UptimeKumaApi(object):
""" """
r = self._get_event_data(Event.PROXY_LIST) r = self._get_event_data(Event.PROXY_LIST)
int_to_bool(r, ["auth", "active", "default", "applyExisting"]) int_to_bool(r, ["auth", "active", "default", "applyExisting"])
parse_proxy_protocol(r)
return r return r
def get_proxy(self, id_: int) -> dict: def get_proxy(self, id_: int) -> dict:
@ -1727,7 +1768,7 @@ class UptimeKumaApi(object):
'id': 1, 'id': 1,
'lastUpdatedDate': None, 'lastUpdatedDate': None,
'pin': 1, 'pin': 1,
'style': 'danger', 'style': <IncidentStyle.DANGER: 'danger'>,
'title': 'title 1' 'title': 'title 1'
}, },
'maintenanceList': [], 'maintenanceList': [],
@ -1769,6 +1810,7 @@ class UptimeKumaApi(object):
"publicGroupList": r2["publicGroupList"], "publicGroupList": r2["publicGroupList"],
"maintenanceList": r2["maintenanceList"] "maintenanceList": r2["maintenanceList"]
} }
parse_incident_style(data["incident"])
return data return data
def add_status_page(self, slug: str, title: str) -> dict: def add_status_page(self, slug: str, title: str) -> dict:
@ -1921,7 +1963,7 @@ class UptimeKumaApi(object):
'createdDate': '2022-12-15 16:51:43', 'createdDate': '2022-12-15 16:51:43',
'id': 1, 'id': 1,
'pin': True, 'pin': True,
'style': 'danger', 'style': <IncidentStyle.DANGER: 'danger'>,
'title': 'title 1' 'title': 'title 1'
} }
""" """
@ -1932,6 +1974,7 @@ class UptimeKumaApi(object):
} }
r = self._call('postIncident', (slug, incident))["incident"] r = self._call('postIncident', (slug, incident))["incident"]
self.save_status_page(slug) self.save_status_page(slug)
parse_incident_style(r)
return r return r
def unpin_incident(self, slug: str) -> dict: def unpin_incident(self, slug: str) -> dict:
@ -1974,7 +2017,7 @@ class UptimeKumaApi(object):
'monitor_id': 1, 'monitor_id': 1,
'msg': '', 'msg': '',
'ping': 10.5, 'ping': 10.5,
'status': True, 'status': <MonitorStatus.UP: 1>,
'time': '2023-05-01 17:22:20.289' 'time': '2023-05-01 17:22:20.289'
}, },
{ {
@ -1985,7 +2028,7 @@ class UptimeKumaApi(object):
'monitor_id': 1, 'monitor_id': 1,
'msg': '', 'msg': '',
'ping': 10.7, 'ping': 10.7,
'status': True, 'status': <MonitorStatus.UP: 1>,
'time': '2023-05-01 17:23:20.349' 'time': '2023-05-01 17:23:20.349'
} }
] ]
@ -2015,7 +2058,7 @@ class UptimeKumaApi(object):
'monitorID': 1, 'monitorID': 1,
'msg': '', 'msg': '',
'ping': 10.5, 'ping': 10.5,
'status': True, 'status': <MonitorStatus.UP: 1>,
'time': '2023-05-01 17:22:20.289' 'time': '2023-05-01 17:22:20.289'
} }
] ]
@ -2869,7 +2912,7 @@ class UptimeKumaApi(object):
[ [
{ {
'dockerDaemon': '/var/run/docker.sock', 'dockerDaemon': '/var/run/docker.sock',
'dockerType': 'socket', 'dockerType': <DockerType.SOCKET: 'socket'>,
'id': 1, 'id': 1,
'name': 'name 1', 'name': 'name 1',
'userID': 1 'userID': 1
@ -2877,6 +2920,7 @@ class UptimeKumaApi(object):
] ]
""" """
r = self._get_event_data(Event.DOCKER_HOST_LIST) r = self._get_event_data(Event.DOCKER_HOST_LIST)
parse_docker_type(r)
return r return r
def get_docker_host(self, id_: int) -> dict: def get_docker_host(self, id_: int) -> dict:
@ -3019,7 +3063,7 @@ class UptimeKumaApi(object):
"id": 1, "id": 1,
"title": "title", "title": "title",
"description": "description", "description": "description",
"strategy": "single", "strategy": <MaintenanceStrategy.SINGLE: 'single'>,
"intervalDay": 1, "intervalDay": 1,
"active": true, "active": true,
"dateRange": [ "dateRange": [
@ -3052,7 +3096,9 @@ class UptimeKumaApi(object):
} }
] ]
""" """
return list(self._get_event_data(Event.MAINTENANCE_LIST).values()) r = list(self._get_event_data(Event.MAINTENANCE_LIST).values())
parse_maintenance_strategy(r)
return r
def get_maintenance(self, id_: int) -> dict: def get_maintenance(self, id_: int) -> dict:
""" """
@ -3070,7 +3116,7 @@ class UptimeKumaApi(object):
"id": 1, "id": 1,
"title": "title", "title": "title",
"description": "description", "description": "description",
"strategy": "single", "strategy": <MaintenanceStrategy.SINGLE: 'single'>,
"intervalDay": 1, "intervalDay": 1,
"active": true, "active": true,
"dateRange": [ "dateRange": [
@ -3103,7 +3149,9 @@ class UptimeKumaApi(object):
"status": "ended" "status": "ended"
} }
""" """
return self._call('getMaintenance', id_)["maintenance"] r = self._call('getMaintenance', id_)["maintenance"]
parse_maintenance_strategy(r)
return r
@append_docstring(maintenance_docstring("add")) @append_docstring(maintenance_docstring("add"))
def add_maintenance(self, **kwargs) -> dict: def add_maintenance(self, **kwargs) -> dict: