DGNum/uptime-kuma-api
6
0
Fork 1
Code Issues 17 Pull requests 5 Projects Releases 24 Packages Wiki Activity Actions

test: create objects with all available arguments

Browse source
This commit is contained in:
lucasheld 2022-09-17 12:24:08 +02:00
parent de38586bf5
commit 928462c6b2
7 changed files with 136 additions and 36 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

75
tests/test_monitor.py
View file

@ -1,17 +1,29 @@
import unittest
from packaging.version import parse as parse_version
from uptime_kuma_api import UptimeKumaException, MonitorType
from uptime_kuma_api import UptimeKumaException, MonitorType, AuthMethod
from uptime_kuma_test_case import UptimeKumaTestCase
class TestMonitor(UptimeKumaTestCase):
def test_monitor(self):
notification_id_1 = self.add_notification()
notification_id_2 = self.add_notification()
expected_monitor = {
"type": MonitorType.HTTP,
"name": "monitor 1",
"interval": 60,
"retryInterval": 60,
"maxretries": 0,
"notificationIDList": [notification_id_1, notification_id_2],
"upsideDown": False,
"url": "http://127.0.0.1"
}
if parse_version(self.api.version) >= parse_version("1.18"):
expected_monitor.update({
"resendInterval": 0
})
# add monitor
r = self.api.add_monitor(**expected_monitor)
@ -60,18 +72,57 @@ class TestMonitor(UptimeKumaTestCase):
self.assertEqual(r["msg"], "Added Successfully.")
monitor_id = r["monitorID"]
monitor = self.api.get_monitor(monitor_id)
self.compare(monitor, expected_monitor)
r = self.api.edit_monitor(monitor_id, **expected_monitor)
self.assertEqual(r["msg"], "Saved.")
monitor = self.api.get_monitor(monitor_id)
self.compare(monitor, expected_monitor)
monitor = self.api.get_monitor(monitor_id)
self.compare(monitor, expected_monitor)
return monitor
def test_monitor_type_http(self):
proxy_id = self.add_proxy()
json_data = '{"key": "value"}'
expected_monitor = {
"type": MonitorType.HTTP,
"name": "monitor 1",
"url": "http://127.0.0.1",
"expiryNotification": False,
"ignoreTls": False,
"maxredirects": 10,
"accepted_statuscodes": ["200-299"],
"proxyId": proxy_id,
"method": "GET",
"body": json_data,
"headers": json_data
"headers": json_data,
"authMethod": AuthMethod.NONE
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_auth_method(self):
for auth_method in [AuthMethod.HTTP_BASIC, AuthMethod.NTLM]:
expected_monitor = {
"type": MonitorType.HTTP,
"name": "monitor 1",
"url": "http://127.0.0.1",
"authMethod": auth_method,
"basic_auth_user": "auth user",
"basic_auth_pass": "auth pass",
}
self.do_test_monitor_type(expected_monitor)
expected_monitor = {
"type": MonitorType.HTTP,
"name": "monitor 1",
"url": "http://127.0.0.1",
"authMethod": AuthMethod.NTLM,
"authDomain": "auth domain",
"authWorkstation": "auth workstation",
}
self.do_test_monitor_type(expected_monitor)
@ -107,7 +158,8 @@ class TestMonitor(UptimeKumaTestCase):
"name": "monitor 1",
"hostname": "127.0.0.1",
"port": 8888,
"dns_resolve_server": "1.1.1.1"
"dns_resolve_server": "1.1.1.1",
"dns_resolve_type": "A"
}
self.do_test_monitor_type(expected_monitor)
@ -136,7 +188,10 @@ class TestMonitor(UptimeKumaTestCase):
"name": "monitor 1",
"hostname": "127.0.0.1",
"port": 8888,
"mqttTopic": "test"
"mqttUsername": "mqtt username",
"mqttPassword": "mqtt password",
"mqttTopic": "mqtt topic",
"mqttSuccessMessage": "mqtt success message"
}
self.do_test_monitor_type(expected_monitor)
@ -190,18 +245,6 @@ class TestMonitor(UptimeKumaTestCase):
}
self.do_test_monitor_type(expected_monitor)
def test_notification_id_list(self):
monitor_id = self.add_monitor()
notification_id = self.add_notification()
expected_monitor = self.api.get_monitor(monitor_id)
expected_monitor["notificationIDList"] = [notification_id]
r = self.api.edit_monitor(id_=monitor_id, **expected_monitor)
self.assertEqual(r["msg"], "Saved.")
monitor = self.api.get_monitor(monitor_id)
self.compare(monitor, expected_monitor)
if __name__ == '__main__':
unittest.main()

2
tests/test_notification.py
View file

@ -8,7 +8,7 @@ class TestNotification(UptimeKumaTestCase):
def test_notification(self):
expected_notification = {
"name": "notification 1",
"default": True,
"isDefault": True,
"applyExisting": True,
"type": "PushByTechulus",
"pushAPIKey": "123456789"

12
tests/test_proxy.py
View file

@ -1,20 +1,24 @@
import unittest
from uptime_kuma_api import UptimeKumaException
from uptime_kuma_api import UptimeKumaException, ProxyProtocol
from uptime_kuma_test_case import UptimeKumaTestCase
class TestProxy(UptimeKumaTestCase):
def test_proxy(self):
expected_proxy = {
"protocol": "http",
"protocol": ProxyProtocol.HTTP,
"host": "127.0.0.1",
"port": 8080,
"active": True
"auth": True,
"username": "username",
"password": "password",
"active": True,
"default": False
}
# add proxy
r = self.api.add_proxy(**expected_proxy)
r = self.api.add_proxy(applyExisting=False, **expected_proxy)
self.assertEqual(r["msg"], "Saved")
proxy_id = r["id"]

28
tests/test_settings.py
View file

@ -1,18 +1,36 @@
import unittest
import json
import unittest
from packaging.version import parse as parse_version
from uptime_kuma_test_case import UptimeKumaTestCase
class TestSettings(UptimeKumaTestCase):
def test_settings(self):
settings_before = self.api.get_settings()
expected_settings = {
"checkUpdate": False,
"checkBeta": False,
"keepDataPeriodDays": 180,
"entryPage": "dashboard",
"searchEngineIndex": False,
"primaryBaseURL": "",
"steamAPIKey": "",
"tlsExpiryNotifyDays": [7, 14, 21],
"disableAuth": False
}
expected_check_update = not settings_before.get("checkUpdate", True)
self.api.set_settings(self.password, checkUpdate=expected_check_update)
if parse_version(self.api.version) >= parse_version("1.18"):
expected_settings.update({
"trustProxy": False
})
# set settings
r = self.api.set_settings(self.password, **expected_settings)
self.assertEqual(r["msg"], "Saved")
# get settings
settings = self.api.get_settings()
self.assertEqual(settings["checkUpdate"], expected_check_update)
self.compare(settings, expected_settings)
def test_change_password(self):
new_password = "321terces"

7
tests/test_status_page.py
View file

@ -13,7 +13,14 @@ class TestStatusPage(UptimeKumaTestCase):
"slug": slug,
"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',

36
tests/uptime_kuma_test_case.py
View file

@ -70,25 +70,45 @@ class UptimeKumaTestCase(unittest.TestCase):
return obj
def add_monitor(self):
r = self.api.add_monitor(type=MonitorType.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
def add_tag(self):
r = self.api.add_tag(name="tag 1", color="#ffffff")
r = self.api.add_tag(
name="tag 1",
color="#ffffff"
)
tag_id = r["id"]
return tag_id
def add_notification(self):
r = self.api.add_notification(name="notification 1", type="PushByTechulus", pushAPIKey="123456789")
r = self.api.add_notification(
name="notification 1",
type="PushByTechulus",
pushAPIKey="123456789"
)
notification_id = r["id"]
return notification_id
def add_proxy(self):
r = self.api.add_proxy(
protocol="http",
host="127.0.0.1",
port=8080,
active=True
)
proxy_id = r["id"]
return proxy_id
def add_docker_host(self):
expected_docker_host = {
"name": "docker host 1",
"dockerType": DockerType.SOCKET
}
r = self.api.add_docker_host(**expected_docker_host)
r = self.api.add_docker_host(
name="docker host 1",
dockerType=DockerType.SOCKET
)
docker_host_id = r["id"]
return docker_host_id

12
uptime_kuma_api/api.py
View file

@ -63,6 +63,14 @@ def _build_notification_data(
applyExisting: bool = False,
**kwargs
):
allowed_kwargs = []
for keys in notification_provider_options.values():
allowed_kwargs.extend(keys)
for key in kwargs.keys():
if key not in allowed_kwargs:
raise TypeError(f"unknown argument '{key}'")
data = {
"name": name,
"type": type,
@ -512,12 +520,12 @@ class UptimeKumaApi(object):
"authWorkstation": authWorkstation,
})
# DNS, PING, STEAM, MQTT
# PORT, PING, DNS, STEAM, MQTT
data.update({
"hostname": hostname,
})
# DNS, STEAM, MQTT
# PORT, DNS, STEAM, MQTT
data.update({
"port": port,
})