reset uptime kuma config after each test
This commit is contained in:
parent
8f108a700b
commit
804bd85909
2 changed files with 30 additions and 16 deletions
|
@ -1,7 +1,7 @@
|
||||||
|
import json
|
||||||
import unittest
|
import unittest
|
||||||
|
|
||||||
from uptime_kuma_api import UptimeKumaApi
|
from uptime_kuma_api import UptimeKumaApi, Event, MonitorType
|
||||||
|
|
||||||
|
|
||||||
token = None
|
token = None
|
||||||
|
|
||||||
|
@ -12,22 +12,32 @@ class UptimeKumaTestCase(unittest.TestCase):
|
||||||
username = "admin"
|
username = "admin"
|
||||||
password = "secret123"
|
password = "secret123"
|
||||||
|
|
||||||
@classmethod
|
def setUp(self):
|
||||||
def setUpClass(cls):
|
self.api = UptimeKumaApi(self.url)
|
||||||
cls.api = UptimeKumaApi(cls.url)
|
|
||||||
|
|
||||||
global token
|
global token
|
||||||
if not token:
|
if not token:
|
||||||
if cls.api.need_setup():
|
if self.api.need_setup():
|
||||||
cls.api.setup(cls.username, cls.password)
|
self.api.setup(self.username, self.password)
|
||||||
r = cls.api.login(cls.username, cls.password)
|
r = self.api.login(self.username, self.password)
|
||||||
token = r["token"]
|
token = r["token"]
|
||||||
|
|
||||||
cls.api.login_by_token(token)
|
self.api.login_by_token(token)
|
||||||
|
|
||||||
@classmethod
|
data = {
|
||||||
def tearDownClass(cls):
|
"version": "1.17.1",
|
||||||
cls.api.disconnect()
|
"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):
|
def compare(self, superset, subset):
|
||||||
self.assertTrue(subset.items() <= superset.items())
|
self.assertTrue(subset.items() <= superset.items())
|
||||||
|
|
|
@ -366,6 +366,7 @@ def _check_arguments_proxy(kwargs):
|
||||||
|
|
||||||
class UptimeKumaApi(object):
|
class UptimeKumaApi(object):
|
||||||
def __init__(self, url):
|
def __init__(self, url):
|
||||||
|
self.url = url
|
||||||
self.sio = socketio.Client()
|
self.sio = socketio.Client()
|
||||||
|
|
||||||
self._event_data: dict = {
|
self._event_data: dict = {
|
||||||
|
@ -396,7 +397,7 @@ class UptimeKumaApi(object):
|
||||||
self.sio.on(Event.INFO, self._event_info)
|
self.sio.on(Event.INFO, self._event_info)
|
||||||
self.sio.on(Event.CERT_INFO, self._event_cert_info)
|
self.sio.on(Event.CERT_INFO, self._event_cert_info)
|
||||||
|
|
||||||
self.connect(url)
|
self.connect()
|
||||||
|
|
||||||
def _get_event_data(self, event):
|
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]
|
monitor_events = [Event.AVG_PING, Event.UPTIME, Event.HEARTBEAT_LIST, Event.IMPORTANT_HEARTBEAT_LIST, Event.CERT_INFO, Event.HEARTBEAT]
|
||||||
|
@ -489,9 +490,12 @@ class UptimeKumaApi(object):
|
||||||
|
|
||||||
# connection
|
# connection
|
||||||
|
|
||||||
def connect(self, url: str):
|
def connect(self):
|
||||||
url = url.rstrip("/")
|
url = self.url.rstrip("/")
|
||||||
|
try:
|
||||||
self.sio.connect(f'{url}/socket.io/')
|
self.sio.connect(f'{url}/socket.io/')
|
||||||
|
except:
|
||||||
|
print("")
|
||||||
|
|
||||||
def disconnect(self):
|
def disconnect(self):
|
||||||
self.sio.disconnect()
|
self.sio.disconnect()
|
||||||
|
|
Loading…
Reference in a new issue